KAFKA-7632: Support Compression Level #10826

dongjinleekr · 2021-06-05T16:32:13Z

Since I reworked KIP-390 from scratch, here I open a new PR.

Committer Checklist (excluded from commit message)

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

dongjinleekr · 2021-06-05T16:34:12Z

@ijuma @guozhangwang Could you have a look when you are free? I refined the original work with a benchmark with a real-world dataset. As you can see in the updated KIP, this option can improve the producer's performance significantly. 🙏

clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java

clients/src/main/java/org/apache/kafka/common/record/CompressionConfig.java

core/src/main/scala/kafka/server/KafkaConfig.scala

kkonstantine · 2021-07-08T05:45:16Z

This PR implements KIP-390 but is still open and has several conflicts. Unless I hear from you by EOD Thursday, July 8 (pacific time), I'll remove it from the KIP list for 3.0

dongjinleekr · 2021-07-08T08:53:06Z

@kkonstantine Here it is - I rebased it onto the latest trunk. Could anyone review this PR? 🙏

kkonstantine · 2021-07-08T22:13:43Z

@tombentley do you think you'll be able to finish the review soon? If we had an approval I'd be willing to make an exception and consider this feature for inclusion to AK 3.0.

mimaison

Thanks for the PR!

I've only looked at the changes in clients so far. It looks good overall, I've left a few questions. Can you also rebase on trunk?

mimaison · 2022-04-14T13:50:54Z

clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java

+    /** <code>compression.level</code> */
+    public static final String COMPRESSION_LEVEL_CONFIG = "compression.level";
+    private static final String COMPRESSION_LEVEL_DOC = "The compression level for all data generated by the producer. The default level and valid value is up to "
+        + "compression.type. (<code>none</code>, <code>snappy</code>: not available. <code>gzip</code>: 1~9. <code>lz4</code>: 1~17. "


Should we include the defaults too?

I think the reason why @dongjinleekr didn't include those is that they may change at the compression library level.

mimaison · 2022-04-14T14:02:29Z

clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java

@@ -523,6 +532,45 @@ boolean idempotenceEnabled() {
        return userConfiguredTransactions || idempotenceEnabled;
    }

+    public CompressionConfig getCompressionConfig(CompressionType compressionType) {
+        if (getString(ProducerConfig.COMPRESSION_LEVEL_CONFIG).isEmpty()) {


Since only 3 cases use the compression level, what about moving the if/else inside each case block?

Suggested change

if (getString(ProducerConfig.COMPRESSION_LEVEL_CONFIG).isEmpty()) {

public CompressionConfig getCompressionConfig(CompressionType compressionType) {

String compressionLevel = getString(ProducerConfig.COMPRESSION_LEVEL_CONFIG);

switch (compressionType) {

case NONE:

return CompressionConfig.NONE;

case GZIP:

if (compressionLevel.isEmpty())

return CompressionConfig.gzip().build();

else

return CompressionConfig.gzip().setLevel(Integer.parseInt(compressionLevel)).build();

case SNAPPY:

return CompressionConfig.snappy().build();

case LZ4:

if (compressionLevel.isEmpty())

return CompressionConfig.lz4().build();

else

return CompressionConfig.lz4().setLevel(Integer.parseInt(compressionLevel)).build();

case ZSTD:

if (compressionLevel.isEmpty())

return CompressionConfig.zstd().build();

else

return CompressionConfig.zstd().setLevel(Integer.parseInt(compressionLevel)).build();

default:

throw new IllegalArgumentException("Unknown compression type: " + compressionType.name);

}

}

Why don't we push some of these things to the enum itself? CompressionType.config(String level) or something like that.

The reason why I did not choose a simple approach like CompressionType.config(String level) (as @ijuma pointed out) here is simple: To make room for supporting per-codec configuration in the future, like KIP-780.

The reason why I chose to use compression.level without default value (as @mimaison pointed out) is also simple: at the point of when KIP-390 is passed, the proposal included global configuration option (i.e., compression.level). But, as I work on KIP-780 and running an in-house preview, I found that per-codec configuration (i.e., compression.{gzip|lz4|zstd}.level) is much more preferable for the following reasons:

1. Easy to switch the compression type

For example, {compression.type=zstd,compression.level=10} works but {compression.type=gzip,compression.level=10} does not, since gzip only supports the level of 1 to 9. In other words, compression.type is so error-prone when the user tries to switch the codec. (@junrao indicated a similar problem: #1 #2)

If we limit the scope of per-codec options like compression.gzip.level, we can easily switch the whole options by changing compression.type only. Moreover, this approach is consistent with additional compression options like compression.gzip.buffer.

2. Can provide a default level.

As @ijuma pointed out, we can not provide a default compression level with compression.level; but with compression.{gzip|lz4|zstd}.level, we can.

For these reasons, I run the overall benchmarks in here with per-config options (see 'Final Notes' section.) If the reviewers prefer this scheme, I will happily change it.

Thanks @dongjinleekr for the explanations. My suggested change was logically identical, it's just a bit shorter.

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

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

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

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

ijuma · 2022-04-14T14:24:05Z

@mimaison fyi, I think we need more votes in the KIP thread.
Edit: Nevermind, this KIP (KIP-390) was adopted. I mixed it up with KIP-780.

mimaison · 2022-04-14T14:28:19Z

As far as I can tell this is implementing KIP-390 and it received 3 binding votes: https://lists.apache.org/thread/wnd6ky3kjzv9gkzc39qy5gg2pp20ovs1

ijuma · 2022-04-14T14:28:39Z

@mimaison yeah, I just edited my message. :)

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

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

junrao

@dongjinleekr : Thanks for the PR and sorry for the late review. Made a pass of all non-testing files. A few comments below.

junrao · 2022-04-13T18:28:30Z

clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java

@@ -312,6 +320,7 @@ public class ProducerConfig extends AbstractConfig {
                                        Importance.LOW,
                                        ACKS_DOC)
                                .define(COMPRESSION_TYPE_CONFIG, Type.STRING, "none", Importance.HIGH, COMPRESSION_TYPE_DOC)
+                                .define(COMPRESSION_LEVEL_CONFIG, Type.STRING, "", Importance.MEDIUM, COMPRESSION_LEVEL_DOC)


The KIP says the default value is null.

Could you kindly review the discussion here?

clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java

junrao · 2022-04-13T18:42:07Z

clients/src/main/java/org/apache/kafka/common/record/CompressionConfig.java

+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.record;


Hmm, it seems this class should be in common.compress package instead of here?

At first I tried so, but it has the following problems:

Consistency with org.apache.kafka.common.record.CompressionType.

CompressionConfig uses org.apache.kafka.common.record.(MemoryRecordsBuilder|RecordBatch) but only the imports from org.apache.kafka.common.record to org.apache.kafka.common.compress are allowed.

clients/src/main/java/org/apache/kafka/common/record/CompressionConfig.java

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

core/src/main/scala/kafka/log/LogConfig.scala

junrao · 2022-04-14T16:44:08Z

core/src/main/scala/kafka/log/LogConfig.scala

@@ -96,6 +97,7 @@ case class LogConfig(props: java.util.Map[_, _], overriddenConfigs: Set[String]
  val uncleanLeaderElectionEnable = getBoolean(LogConfig.UncleanLeaderElectionEnableProp)
  val minInSyncReplicas = getInt(LogConfig.MinInSyncReplicasProp)
  val compressionType = getString(LogConfig.CompressionTypeProp).toLowerCase(Locale.ROOT)
+  val compressionLevel = getString(LogConfig.CompressionLevelProp)


With topic level overriding, things can be a bit tricky. Suppose that a user sets up a broker level compression type and compression level, but only overrides the compression type for a topic. With this logic, the topic will pick up the broker level compression level, which could be intended for a different compression type.

ditto: here

core/src/main/scala/kafka/log/UnifiedLog.scala

junrao · 2022-04-14T16:58:12Z

core/src/main/scala/kafka/log/LogConfig.scala

+  /**
+   * Returns per-codec [[CompressionConfig]] for given `compressionType`.
+   */
+  private def getCompressionConfig(compressionType: CompressionType): CompressionConfig = {


Another thing is that if we set a topic or broker level compression type to be the same as the producer, but with a different compression level, whether we should trigger broker side recompression. Currently, we don't do recompression. Should we do it? If so, is it possible to know the producer compression level?

No, we can't trigger broker side recompression in these scenario since there is no field for compression level in RecordBatch. (Oh, this should be also addressed in the KIP.)

raft/src/main/java/org/apache/kafka/snapshot/SnapshotWriter.java

…eConfig, GzipConfig, SnappyConfig, LZ4Config, ZstdConfig. - Now, each compression codec can have independent config entries. - Make MemoryRecordsBuilder, KafkaRaftClient, SnapshotWriter to use CompressionConfig instead of CompressionType. - Make LogAppendInfo to use (CompressionType, CompressionConfig) for compression source and target. - Add ProducerConfig#getCompressionConfig: parses compression configs and returns per-codec CompressionConfig from given configurations. - Add LogConfig#getCompressionConfig: parses compression configs and returns per-codec CompressionConfig from given configurations. - Separate CompressedRecordBatchValidationBenchmark into [Gzip, Snappy, LZ4, Zstd]CompressedRecordBatchValidationBenchmark

1. Remove unused constructor in KafkaLZ4BlockOutputStream. (and now it is used for the compression level instead.) 2. Add KafkaLZ4BlockOutputStream.[MIN_COMPRESSION_LEVEL, MAX_COMPRESSION_LEVEL, DEFAULT_COMPRESSION_LEVEL] (from LZ4Factory#highCompressor and LZ4Constants.[DEFAULT_COMPRESSION_LEVEL, MAX_COMPRESSION_LEVEL]

- Change scope of 'ProducerConfig#getCompressionConfig(CompressionType)' into package. - Revert import order + Fix parameter name in RecordAccumulator. - Add Javadoc params to CompressionConfig#wrapForInput. - Fix Javadoc param in RecordsSnapshotWriter. - Add '()' to 'CompressionCodec.Builder#build' call in scala.

dongjinleekr · 2022-07-08T20:03:13Z

Sorry for being late. Rebased onto the latest trunk, with resolving the issues raised by @mimaison @ijuma @junrao.

mimaison

Thanks @dongjinleekr for the updates. I've made another pass and left a few minor comments.

mimaison · 2022-08-08T09:24:02Z

clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java

@@ -523,6 +532,45 @@ boolean idempotenceEnabled() {
        return userConfiguredTransactions || idempotenceEnabled;
    }

+    public CompressionConfig getCompressionConfig(CompressionType compressionType) {
+        if (getString(ProducerConfig.COMPRESSION_LEVEL_CONFIG).isEmpty()) {


Thanks @dongjinleekr for the explanations. My suggested change was logically identical, it's just a bit shorter.

mimaison · 2022-08-08T09:35:38Z

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

+     * <code>compression.level</code>
+     */
+    public static final String COMPRESSION_LEVEL_CONFIG = "compression.level";
+    public static final String COMPRESSION_LEVEL_DOC = "The compression level for all data generated by the producer. The default level and valid value is up to "


What about:

The compression level to use when compressing records broker side. This is only used if compression.type is set to gzip, lz4 or zstd. The default level and valid value is up to...

mimaison · 2022-08-08T10:11:33Z

clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java

@@ -289,8 +290,7 @@ private void testAppendLargeOldMessageFormat(CompressionType compressionType) th
    public void testLinger() throws Exception {
        int lingerMs = 10;
        RecordAccumulator accum = createTestRecordAccumulator(
-                1024 + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10 * 1024, CompressionType.NONE, lingerMs);
-        accum.append(topic, partition1, 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs, false, time.milliseconds(), cluster);


Don't we need this line for the test to pass?

mimaison · 2022-08-08T10:13:41Z

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

+        private int level = DEFAULT_COMPRESSION_LEVEL;
+
+        public Builder level(int level) {
+            if (MAX_COMPRESSION_LEVEL < level) {


Should we also check the min level?

mimaison · 2022-08-08T10:15:16Z

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

+
+        public Builder level(int level) {
+            if (level < Lz4OutputStream.MIN_COMPRESSION_LEVEL || Lz4OutputStream.MAX_COMPRESSION_LEVEL < level) {
+                throw new IllegalArgumentException("lz4 doesn't support given compression level: " + level);


Can we include the min and max values in the error message?

mimaison · 2022-08-08T10:17:12Z

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

-                            for (boolean close : Arrays.asList(false, true))
-                                arguments.add(Arguments.of(new Args(broken, ignore, blockChecksum, close, payload)));
+                        // Available levels: 1, 2, ... 17.
+                        for (int compressionLevel = 1; compressionLevel < 18; ++compressionLevel)


Should we use Lz4OutputStream.MIN_COMPRESSION_LEVEL and MAX_COMPRESSION_LEVEL rather than hardcoding the levels?

mimaison · 2022-08-08T10:38:44Z

core/src/test/scala/other/kafka/TestLinearWriteSpeed.scala

    val rand = new Random
    rand.nextBytes(buffer.array)
    val numMessages = bufferSize / (messageSize + Records.LOG_OVERHEAD)
    val createTime = System.currentTimeMillis
+    val properties = new Properties()
+    properties ++= extraProps
+    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000")


Ah right this is only created in order to call getCompressionConfig. For a second I was wondering where localhost:9000 came from.

clolov · 2023-01-23T11:24:37Z

Hello @dongjinleekr! What is the current state of this pull request?

ozturkberkay · 2023-10-31T09:37:45Z

Hi, I was wondering if this PR would be prioritized in the foreseeable future given there has been a lot of work done already

junrao · 2023-11-06T19:01:34Z

One discussion of the PR is around choosing the default value for compression.level (#10826 (comment)). This is tricky since it's specific to the compression type. I am wondering if it's better to adjust the KIP to support compression.{gzip|lz4|zstd}.level instead. This will make it more aligned with the future KIP https://cwiki.apache.org/confluence/display/KAFKA/KIP-780%3A+Support+fine-grained+compression+options#KIP780:Supportfinegrainedcompressionoptions-Zstd.1.

mimaison · 2023-11-08T13:56:42Z

Yes it would be good to get this over the line. @dongjinleekr Do you think you'll have time to update it? or would you need some help?

mimaison · 2024-02-07T10:19:50Z

I updated the KIP based on Jun's suggestion and sent an update to the VOTE thread.

mimaison · 2024-04-10T17:42:24Z

I opened #15516 to implement this KIP.

mimaison · 2024-06-07T11:13:59Z

Closing this PR since #15516 has been merged.

dongjinleekr · 2024-07-15T11:38:16Z

@junrao @mimaison @clolov @ozturkberkay

I greatly appreciate you finalizing this PR! After changing jobs, I could not get involved with the Kafka community since I had to focus on our in-house fork, aka Navercorp (@naver) Kafka.

Now, I just restarted KIP-780 and am testing it with our in-house tiered storage plugin. As soon as I complete the work, I will re-open the PR. Thank you again, and stay tuned! 🙇‍♂️

dongjinleekr mentioned this pull request Jun 5, 2021

KAFKA-7632: Allow fine-grained configuration for compression #5927

Closed

3 tasks

tombentley reviewed Jun 9, 2021

View reviewed changes

dongjinleekr force-pushed the feature/KAFKA-7632 branch from 6a63961 to baa5914 Compare June 16, 2021 08:58

dongjinleekr force-pushed the feature/KAFKA-7632 branch from baa5914 to 1899e8e Compare July 8, 2021 08:17

dongjinleekr force-pushed the feature/KAFKA-7632 branch from 1899e8e to ff1a3ec Compare October 10, 2021 11:52

dongjinleekr force-pushed the feature/KAFKA-7632 branch from ff1a3ec to 46bf0ac Compare October 30, 2021 13:09

mimaison reviewed Apr 14, 2022

View reviewed changes

ijuma reviewed Apr 14, 2022

View reviewed changes

clients/src/main/java/org/apache/kafka/common/compress/GzipConfig.java Outdated Show resolved Hide resolved

ijuma reviewed Apr 14, 2022

View reviewed changes

clients/src/main/java/org/apache/kafka/common/compress/GzipOutputStream.java Show resolved Hide resolved

junrao reviewed Apr 14, 2022

View reviewed changes

dongjinleekr added 9 commits July 8, 2022 19:28

Make GzipConfig to support compression level

d493d41

Make ZstdConfig to support compression level

29dbaf2

Add compression.level configuration

89c1182

Rename class name: KafkaLZ4BlockOutputStream → Lz4OutputStream

b334052

Change method names: getType → type. setLevel → level.

8ef745b

Remove unused constructors in Lz4OutputStream

2c96abf

dongjinleekr force-pushed the feature/KAFKA-7632 branch from 46bf0ac to 2c96abf Compare July 8, 2022 20:02

mimaison reviewed Aug 8, 2022

View reviewed changes

mimaison mentioned this pull request Apr 10, 2024

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

Merged

3 tasks

mimaison closed this Jun 7, 2024

dongjinleekr deleted the feature/KAFKA-7632 branch July 15, 2024 11:38

-        if (getString(ProducerConfig.COMPRESSION_LEVEL_CONFIG).isEmpty()) {
+        public CompressionConfig getCompressionConfig(CompressionType compressionType) {
+        String compressionLevel = getString(ProducerConfig.COMPRESSION_LEVEL_CONFIG);
+        switch (compressionType) {
+            case NONE:
+                return CompressionConfig.NONE;
+            case GZIP:
+                if (compressionLevel.isEmpty())
+                    return CompressionConfig.gzip().build();
+                else
+                    return CompressionConfig.gzip().setLevel(Integer.parseInt(compressionLevel)).build();
+            case SNAPPY:
+                return CompressionConfig.snappy().build();
+            case LZ4:
+                if (compressionLevel.isEmpty())
+                    return CompressionConfig.lz4().build();
+                else
+                    return CompressionConfig.lz4().setLevel(Integer.parseInt(compressionLevel)).build();
+            case ZSTD:
+                if (compressionLevel.isEmpty())
+                    return CompressionConfig.zstd().build();
+                else
+                    return CompressionConfig.zstd().setLevel(Integer.parseInt(compressionLevel)).build();
+            default:
+                throw new IllegalArgumentException("Unknown compression type: " + compressionType.name);
+        }
+    }

KAFKA-7632: Support Compression Level #10826

KAFKA-7632: Support Compression Level #10826

Conversation

dongjinleekr commented Jun 5, 2021

Committer Checklist (excluded from commit message)

dongjinleekr commented Jun 5, 2021

kkonstantine commented Jul 8, 2021

dongjinleekr commented Jul 8, 2021

kkonstantine commented Jul 8, 2021

mimaison left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ijuma Apr 14, 2022 • edited Loading

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

1. Easy to switch the compression type

2. Can provide a default level.

Choose a reason for hiding this comment

ijuma commented Apr 14, 2022 • edited Loading

mimaison commented Apr 14, 2022

ijuma commented Apr 14, 2022

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

dongjinleekr Jul 8, 2022 • edited Loading

Choose a reason for hiding this comment

dongjinleekr commented Jul 8, 2022

mimaison 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

clolov commented Jan 23, 2023

ozturkberkay commented Oct 31, 2023

junrao commented Nov 6, 2023

mimaison commented Nov 8, 2023

mimaison commented Feb 7, 2024

mimaison commented Apr 10, 2024

mimaison commented Jun 7, 2024

dongjinleekr commented Jul 15, 2024

ijuma Apr 14, 2022 •

edited

Loading

ijuma commented Apr 14, 2022 •

edited

Loading

dongjinleekr Jul 8, 2022 •

edited

Loading