[LI-HOTFIX] make GZIP decompression use BufferSupplier (#158)

EXIT_CRITERIA = KAFKA-12605
linkedin · May 11, 2021 · 074c432 · 074c432
1 parent 0d2abca
commit 074c432
Show file tree

Hide file tree

Showing 6 changed files with 1,229 additions and 2 deletions.
diff --git a/clients/src/main/java/org/apache/kafka/common/record/CompressionType.java b/clients/src/main/java/org/apache/kafka/common/record/CompressionType.java
@@ -66,8 +66,18 @@ public InputStream wrapForInput(ByteBuffer buffer, byte messageVersion, BufferSu
                 // Set output buffer (uncompressed) to 16 KB (none by default) and input buffer (compressed) to
                 // 8 KB (0.5 KB by default) to ensure reasonable performance in cases where the caller reads a small
                 // number of bytes (potentially a single byte)
-                return new BufferedInputStream(new GZIPInputStream(new ByteBufferInputStream(buffer), 8 * 1024),
-                        16 * 1024);
+                if (useBufferPoolForGzip) {
+                    return new KafkaBufferedInputStream(
+                            new KafkaGZIPInputStream(
+                                    new ByteBufferInputStream(buffer), decompressionBufferSupplier, 8 * 1024
+                            ),
+                            decompressionBufferSupplier,
+                            16 * 1024
+                    );
+                } else {
+                    return new BufferedInputStream(new GZIPInputStream(new ByteBufferInputStream(buffer), 8 * 1024),
+                            16 * 1024);
+                }
             } catch (Exception e) {
                 throw new KafkaException(e);
             }
@@ -147,6 +157,8 @@ public InputStream wrapForInput(ByteBuffer buffer, byte messageVersion, BufferSu
         }
     };
 
+    public static volatile boolean useBufferPoolForGzip = false;
+
     public final int id;
     public final String name;
     public final float rate;