KAFKA-7632: Support Compression Levels (KIP-390) #15516

mimaison · 2024-03-11T18:40:17Z

Based on #10826 with updates to match the recent amends we made to KIP-390.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

mimaison · 2024-04-10T17:43:31Z

@divijvaidya It seems you've done a bit of work around compression in the past. Can you take a look? Thanks

Co-authored-by: Lee Dongjin <dongjin@apache.org>

showuon

Thanks for the PR. Left some comments.

showuon · 2024-04-25T11:47:27Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java

@@ -189,7 +189,7 @@ public RecordAccumulator(LogContext logContext,
                             BufferPool bufferPool) {
        this(logContext,
            batchSize,
-            compression,
+                compression,


nit: wrong indent

showuon · 2024-04-25T11:49:04Z

clients/src/main/java/org/apache/kafka/common/compress/Compression.java

+     * Wrap bufferStream with an OutputStream that will compress data with this CompressionType.
+     * Note: Unlike {@link #wrapForInput}, this cannot take {@link ByteBuffer}s directly.
+     * Currently, MemoryRecordsBuilder writes to the underlying buffer in the given {@link ByteBufferOutputStream} after the compressed data has been written.
+     * In the event that the buffer needs to be expanded while writing the data, access to the underlying buffer needs to be preserved.
+     */


Please add @param @return .

showuon · 2024-04-25T11:49:11Z

clients/src/main/java/org/apache/kafka/common/compress/Compression.java

+     * @param buffer The {@link ByteBuffer} instance holding the data to decompress.
+     * @param messageVersion The record format version to use.
+     * @param decompressionBufferSupplier The supplier of ByteBuffer(s) used for decompression if supported.
+     * For small record batches, allocating a potentially large buffer (64 KB for LZ4)


miss @return

indentation

showuon · 2024-04-25T11:59:57Z

clients/src/main/java/org/apache/kafka/common/compress/Lz4BlockOutputStream.java

+        /*
+         * lz4-java provides two types of compressors; fastCompressor, which requires less memory but fast compression speed (with default compression level only),
+         * and highCompressor which requires more memory and slower speed but compresses more efficiently (with various compression level).
+         *
+         * For backward compatibility, Lz4BlockOutputStream uses fastCompressor with default compression level but, with the other level, it uses highCompressor.
+         */
+        compressor = level == Lz4Compression.DEFAULT_LEVEL ? LZ4Factory.fastestInstance().fastCompressor() : LZ4Factory.fastestInstance().highCompressor(level);


Should we add this comment in the lz4.level config description? I think users also need to know it.

showuon · 2024-04-25T12:02:19Z

clients/src/main/java/org/apache/kafka/common/compress/Lz4BlockOutputStream.java

-    public static final int BLOCKSIZE_256KB = 5;
-    public static final int BLOCKSIZE_1MB = 6;
-    public static final int BLOCKSIZE_4MB = 7;


They can be removed because we have another KIP to address them. Fine.

showuon · 2024-04-25T12:08:48Z

clients/src/test/java/org/apache/kafka/common/compress/GzipCompressionTest.java

+        builder.level(GzipCompression.MIN_LEVEL);
+        builder.level(GzipCompression.MAX_LEVEL);


nit: assertDoesNotThrow

showuon · 2024-04-25T12:13:02Z

clients/src/test/java/org/apache/kafka/common/compress/Lz4CompressionTest.java

+        builder.level(Lz4Compression.MIN_LEVEL);
+        builder.level(Lz4Compression.MAX_LEVEL);


nit: assertDoesNotThrow

showuon · 2024-04-25T12:16:40Z

clients/src/test/java/org/apache/kafka/common/record/DefaultRecordBatchTest.java

        // Buffer containing compressed records to be used for creating zstd-jni stream
-        ByteBuffer recordsBuffer = compressedBuf.duplicate();
+        ByteBuffer recordsBuffer = spy(compressedBuf.duplicate());


Why should we spy this?

showuon · 2024-04-25T12:31:44Z

storage/src/main/java/org/apache/kafka/storage/internals/log/LogValidator.java

            throw new UnsupportedCompressionTypeException("Produce requests to inter.broker.protocol.version < 2.1 broker " +
                "are not allowed to use ZStandard compression");

        // No in place assignment situation 1
-        boolean inPlaceAssignment = sourceCompression == targetCompression;
+        boolean inPlaceAssignment = sourceCompressionType == targetCompression.type();


So we won't do re-compression if only level is different? I didn't see this in KIP. Maybe we should add it?

showuon · 2024-04-25T12:33:33Z

clients/src/main/java/org/apache/kafka/common/config/TopicConfig.java

+    public static final String COMPRESSION_GZIP_LEVEL_DOC = "The compression level to use if " + COMPRESSION_TYPE_CONFIG + " is set to <code>gzip</code>.";
+    public static final String COMPRESSION_LZ4_LEVEL_CONFIG = "compression.lz4.level";
+    public static final String COMPRESSION_LZ4_LEVEL_DOC = "The compression level to use if " + COMPRESSION_TYPE_CONFIG + " is set to <code>lz4</code>.";
+    public static final String COMPRESSION_ZSTD_LEVEL_CONFIG = "compression.zstd.level";
+    public static final String COMPRESSION_ZSTD_LEVEL_DOC = "The compression level to use if " + COMPRESSION_TYPE_CONFIG + " is set to <code>zstd</code>.";


nit: Should we provide the doc link for each compression type? It's hard to know which level means what.

junrao

@mimaison : Thanks for the PR. Made a pass of non-testing files. Left a few comments.

junrao · 2024-05-01T23:04:12Z

clients/src/main/java/org/apache/kafka/common/compress/GzipCompression.java

+            // Set input buffer (uncompressed) to 16 KB (none by default) and output buffer (compressed) to
+            // 8 KB (0.5 KB by default) to ensure reasonable performance in cases where the caller passes a small
+            // number of bytes to write (potentially a single byte)
+            return new BufferedOutputStream(new GzipOutputStream(buffer, 8 * 1024, this.level), 16 * 1024);


this.level => level Ditto in a few other classes.

junrao · 2024-05-01T23:20:31Z

clients/src/main/java/org/apache/kafka/common/compress/Compression.java

+     * @param buffer The {@link ByteBuffer} instance holding the data to decompress.
+     * @param messageVersion The record format version to use.
+     * @param decompressionBufferSupplier The supplier of ByteBuffer(s) used for decompression if supported.
+     * For small record batches, allocating a potentially large buffer (64 KB for LZ4)


indentation

junrao · 2024-05-01T23:22:16Z

clients/src/main/java/org/apache/kafka/common/compress/Compression.java

+    CompressionType type();
+
+    /**
+     * Wrap bufferStream with an OutputStream that will compress data with this CompressionType.


Could we add the javadoc for messageVersion?

junrao · 2024-05-01T23:31:54Z

clients/src/main/java/org/apache/kafka/common/compress/NoCompression.java

+
+    @Override
+    public int hashCode() {
+        return super.hashCode();


Hmm, we redefine equals() such that all objects of type NoCompression are equal, yet they have different hashcode?

junrao · 2024-05-01T23:36:18Z

clients/src/main/java/org/apache/kafka/common/compress/NoCompression.java

    }

+    public static class Builder implements Compression.Builder<NoCompression> {


Should we make the constructor of NoCompression private so that only the builder can be used? Ditto for other compression classes.

junrao · 2024-05-01T23:45:06Z

clients/src/main/java/org/apache/kafka/common/compress/Lz4Compression.java

+
+    public static final int MIN_LEVEL = 1;
+    public static final int MAX_LEVEL = 17;
+    public static final int DEFAULT_LEVEL = 9;


So, every time we update the Lz4 library, we may need to update the above values? We probably want to add a note here.

junrao · 2024-05-02T20:10:01Z

clients/src/main/java/org/apache/kafka/common/compress/ZstdCompression.java

+    }
+
+    // visible for testing
+    public ZstdInputStreamNoFinalizer wrapForZstdInput(ByteBuffer buffer, BufferSupplier decompressionBufferSupplier) throws IOException {


Should we make this static?

junrao · 2024-05-02T22:13:54Z

clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtils.java

            ByteBufferOutputStream out = new ByteBufferOutputStream(512)) {
+


extra new line

junrao · 2024-05-02T22:25:43Z

storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java

@@ -388,6 +402,32 @@ public LogConfig(Map<?, ?> props, Set<String> overriddenConfigs) {
        remoteLogConfig = new RemoteLogConfig(this);
    }

+    private Optional<Compression> getCompression() {
+        BrokerCompressionType brokerCompressionType = BrokerCompressionType.forName(getString(TopicConfig.COMPRESSION_TYPE_CONFIG));


Could we just use compressionType, which is previously set already?

mimaison force-pushed the kip-390 branch 3 times, most recently from df9ca6e to 2f54aac Compare March 25, 2024 08:47

mimaison force-pushed the kip-390 branch 3 times, most recently from d284cd3 to 71d84bd Compare April 10, 2024 09:04

mimaison marked this pull request as ready for review April 10, 2024 10:00

mimaison force-pushed the kip-390 branch from 71d84bd to 6619e40 Compare April 10, 2024 15:09

mimaison mentioned this pull request Apr 10, 2024

KAFKA-7632: Support Compression Level #10826

Open

3 tasks

showuon self-assigned this Apr 11, 2024

KAFKA-7632: Support Compression Levels (KIP-390)

dd728c8

Co-authored-by: Lee Dongjin <dongjin@apache.org>

mimaison force-pushed the kip-390 branch from 6619e40 to dd728c8 Compare April 11, 2024 08:55

showuon reviewed Apr 25, 2024

View reviewed changes

junrao reviewed May 2, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-7632: Support Compression Levels (KIP-390) #15516

KAFKA-7632: Support Compression Levels (KIP-390) #15516

mimaison commented Mar 11, 2024 •

edited

mimaison commented Apr 10, 2024

showuon left a comment

showuon Apr 25, 2024

showuon Apr 25, 2024

showuon Apr 25, 2024

junrao May 1, 2024

showuon Apr 25, 2024

showuon Apr 25, 2024

showuon Apr 25, 2024

showuon Apr 25, 2024

showuon Apr 25, 2024

showuon Apr 25, 2024

showuon Apr 25, 2024

junrao left a comment

junrao May 1, 2024

junrao May 1, 2024

junrao May 1, 2024

junrao May 1, 2024

junrao May 1, 2024

junrao May 1, 2024

junrao May 2, 2024

junrao May 2, 2024

junrao May 2, 2024

		builder.level(GzipCompression.MIN_LEVEL);
		builder.level(GzipCompression.MAX_LEVEL);

		builder.level(Lz4Compression.MIN_LEVEL);
		builder.level(Lz4Compression.MAX_LEVEL);

		}

		public static class Builder implements Compression.Builder<NoCompression> {

		ByteBufferOutputStream out = new ByteBufferOutputStream(512)) {

KAFKA-7632: Support Compression Levels (KIP-390) #15516

Are you sure you want to change the base?

KAFKA-7632: Support Compression Levels (KIP-390) #15516

Conversation

mimaison commented Mar 11, 2024 • edited

Committer Checklist (excluded from commit message)

mimaison commented Apr 10, 2024

showuon left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

junrao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mimaison commented Mar 11, 2024 •

edited