Enable zSTD Compression for Kafka Buffer - Json type by jeffreyAaron · Pull Request #5778 · opensearch-project/data-prepper

jeffreyAaron · 2025-06-13T18:01:32Z

Description

Implements zSTD compression and decompression for the KafkaBuffer class. Most changes are implemented in the CustomKafkaConsumer and CustomKafkaProducer.

Issues Resolved

Resolves #5777

Check List

New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
- New functionality has javadoc added
Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

san81

Nice work. Just left a few comments. Please take a look

san81 · 2025-06-16T06:54:51Z

        implementation platform('com.fasterxml.jackson:jackson-bom:2.17.2')
        implementation platform('org.eclipse.jetty:jetty-bom:9.4.53.v20231009')
        implementation platform('io.micrometer:micrometer-bom:1.10.5')
+        implementation 'com.github.luben:zstd-jni:1.5.7-3'


do you need this in the global build.gradle file as well?

Not needed, I'll remove it.

san81 · 2025-06-16T07:10:39Z

        this.executorService = Executors.newFixedThreadPool(consumers.size(), KafkaPluginThreadFactory.defaultExecutorThreadFactory(MDC_KAFKA_PLUGIN_VALUE));
        consumers.forEach(this.executorService::submit);
-
+        this.serdeFormat = kafkaBufferConfig.getSerdeFormat();


I am not seeing any references to this newly introduced member. Not sure if we need this here?

Doesn't seem necessary, I will remove it.

KarstenSchnitter · 2025-06-16T08:15:34Z

-                    InputStream inputStream = new ByteArrayInputStream((byte[])consumerRecord.value());
+                    InputStream byteInputStream = new ByteArrayInputStream((byte[])consumerRecord.value());
+                    InputStream inputStream;
+                    if (compressionEnabled) {


Kafka supports multiple compression algorithms: lz4, gzip, snappy, zstd. Do you think just using one boolean parameter is sufficient for configuration? There are also parameters for tuning the algorithms.

I think that the compression stage should be invisible to the user and the flag should be removed altogether. The data outputted is not modified in anyway and makes no difference (other than performance) on the user end. I will remove the config flag in a future commit on this PR.

@KarstenSchnitter that's valid point. But we will not be using Kafka's compression in case of Kafka Buffer, we will be using compression outside the kafka so that we can do compression and encryption in that order. We are starting with ZSTD and may add other options in future.

Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

san81 · 2025-06-17T18:39:52Z

+    public KafkaCustomConsumerFactory(SerializationFactory serializationFactory, AwsCredentialsSupplier awsCredentialsSupplier, boolean isCompressionEnabled) {
+        this.serializationFactory = serializationFactory;
+        this.awsCredentialsSupplier = awsCredentialsSupplier;
+    }
+
+


We don't need this method anymore - right?

Not needed, will remove.

Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

kkondaka

Looks good except for the default config. Please test with KMS keys too.

kkondaka · 2025-06-18T23:32:03Z

    }
+
+    public CompressionConfig getCompressionConfig() {
+        if (Objects.isNull(compressionConfig)) {


I don't think we want to change default behavior. We should only enable compression when compression config is explicitly provided in the YAML

kkondaka · 2025-06-18T23:39:13Z

        this.errLogRateLimiter = new LogRateLimiter(2, System.currentTimeMillis());
    }

+    public KafkaCustomConsumer(final KafkaConsumer consumer,


Usually, we do reverse way when multiple constructors are present. Invoke the second constructor from the first by passing null as the compression config.

constructor (...existing args..., compression-arg) { init all the members } constructor(..existing args...) { this(existing args, null); }

- Also address review comments Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

san81

Essential addition to the Kafka buffer. Thanks for adding this.

dlvenable · 2025-06-19T16:23:50Z

@jeffreyAaron , Did you try the existing configuration? It should tell Kafka to use the zstd compression.

buffer:
  kafka:
    producer_properties:
      compression_type: zstd

dlvenable

Thanks @jeffreyAaron for the contribution. I see the goal is to support compress then encrypt. I have some code changes to recommend.

dlvenable · 2025-06-19T16:35:34Z

+import java.util.Map;
+import java.util.stream.Collectors;
+
+public enum CompressionType {


We should extract this code from the s3 sink.

https://github.com/opensearch-project/data-prepper/tree/1368a2151307e07dc4e05400b981d1b3521b1098/data-prepper-plugins/s3-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/s3/compression

It has the interfaces that we need.

The CompressionEngine interface can be moved here:

https://github.com/opensearch-project/data-prepper/tree/bc4039dedeb624f4269b3437d4b3f2cd15e963c4/data-prepper-api/src/main/java/org/opensearch/dataprepper/model/codec

The CompressionType can be consolidated with this: https://github.com/opensearch-project/data-prepper/blob/09a6497b29b4f1b3dbadb2a7b0708b76e4855c28/data-prepper-plugins/common/src/main/java/org/opensearch/dataprepper/plugins/codec/CompressionOption.java

Add a method like this:

public CompressionEngine getCompressionEngine() { return ... }

And all the CompressionEngine implementations can be put in here: https://github.com/opensearch-project/data-prepper/tree/09a6497b29b4f1b3dbadb2a7b0708b76e4855c28/data-prepper-plugins/common/src/main/java/org/opensearch/dataprepper/plugins/codec

This is a good idea. I'll implement these changes.

dlvenable · 2025-06-19T16:37:09Z

    private String customMetricPrefix;

+    @JsonProperty("compression")
+    private CompressionConfig compressionConfig;


Rather than a new config, we should use the existing one (producer_properties/compression_type).

If encryption is not enabled, then let Kafka encrypt.

If encryption is enabled, then do it manually and disable it in Kafka.

Implemented this logic in KafkaBuffer.java in 947f031

- Extract Compression Engine classes - Implement logic to default to Kafka compression if encryption at rest is Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

jeffreyAaron · 2025-06-20T18:43:18Z

@dlvenable Requested changes are implemented. Could you please take a look when you have time?

jeffreyAaron · 2025-06-20T21:19:31Z

@jeffreyAaron , Did you try the existing configuration? It should tell Kafka to use the zstd compression.
buffer:
  kafka:
    producer_properties:
      compression_type: zstd

Yes, this seems to work with noticeable improvement with the compression enabled.

947f031 Should ensure that compression is done before encryption.

dlvenable

Thank you @jeffreyAaron ! This looks good to me.

* Fix import style and integration test fixes Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

…ect#5778) * Implement Kafka buffering compression Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com> Signed-off-by: Jonah Calvo <caljonah@amazon.com>

…roject#5811) * Fix import style and integration test fixes Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com> Signed-off-by: Jonah Calvo <caljonah@amazon.com>

jeffreyAaron requested review from KarstenSchnitter, chenqi0805, dinujoh, dlvenable, engechas, graytaylor0, kkondaka, oeyh, san81, sb2k16 and srikanthjg as code owners June 13, 2025 18:01

jeffreyAaron force-pushed the main branch from 2dccfdd to 3e6d9cd Compare June 13, 2025 18:04

jeffreyAaron marked this pull request as draft June 13, 2025 18:15

jeffreyAaron added 3 commits June 13, 2025 11:28

Implement Kafka Buffering

93327b5

Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

Enable Kafka buffer compression by default

83451fa

Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

Fix bugs that caused test to fail

83b125d

Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

jeffreyAaron force-pushed the main branch from 8d44ee7 to 83b125d Compare June 13, 2025 18:32

Merge branch 'main' into main

f62a0d4

Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

jeffreyAaron marked this pull request as ready for review June 13, 2025 18:38

jeffreyAaron marked this pull request as draft June 13, 2025 19:04

jeffreyAaron marked this pull request as ready for review June 13, 2025 19:35

san81 reviewed Jun 16, 2025

View reviewed changes

KarstenSchnitter reviewed Jun 16, 2025

View reviewed changes

jeffreyAaron added 3 commits June 16, 2025 11:36

Address review comments and remove config flag

309fa97

Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

Update KafkaBufferIT.java

08452f3

Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

Remove unnecessary dependency

410862c

Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

jeffreyAaron force-pushed the main branch from 8732ee3 to 410862c Compare June 16, 2025 22:36

Remove conditionals for compression

543cdf0

Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

jeffreyAaron force-pushed the main branch from 0b581d1 to 543cdf0 Compare June 17, 2025 17:42

san81 reviewed Jun 17, 2025

View reviewed changes

jeffreyAaron force-pushed the main branch from e3424bd to f3940e3 Compare June 17, 2025 23:13

jeffreyAaron added 3 commits June 18, 2025 11:52

Add conditionals for Kafka compression

6dfb13f

Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

Add compression config flag for Kafka Buffer

8ed2350

Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

Add CompressionConfig object

fb299eb

Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

jeffreyAaron force-pushed the main branch from 6bd2dc9 to fb299eb Compare June 18, 2025 21:26

Fix style

5e5f8e1

Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

jeffreyAaron force-pushed the main branch from 97fd765 to 5e5f8e1 Compare June 18, 2025 22:19

Remove unnecessary variable

b495a1e

Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

kkondaka reviewed Jun 18, 2025

View reviewed changes

jeffreyAaron added 2 commits June 18, 2025 17:46

Change default compression behavior

c79f6ed

- Also address review comments Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

Update the tests to match the default compression

48ae5cf

Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

jeffreyAaron force-pushed the main branch from dfe866a to 48ae5cf Compare June 19, 2025 00:56

san81 previously approved these changes Jun 19, 2025

View reviewed changes

dlvenable requested changes Jun 19, 2025

View reviewed changes

Utilize producer compression

947f031

- Extract Compression Engine classes - Implement logic to default to Kafka compression if encryption at rest is Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

jeffreyAaron dismissed san81’s stale review via 947f031 June 20, 2025 06:14

san81 requested a review from dlvenable June 20, 2025 19:44

dlvenable approved these changes Jun 20, 2025

View reviewed changes

san81 approved these changes Jun 20, 2025

View reviewed changes

san81 merged commit 691a737 into opensearch-project:main Jun 20, 2025
5 of 9 checks passed

jeffreyAaron mentioned this pull request Jun 23, 2025

Fix style and build errors from #5778 #5811

Merged

4 tasks

san81 pushed a commit that referenced this pull request Jun 23, 2025

Fix style and build errors from #5778 (#5811)

4dab7ac

* Fix import style and integration test fixes Signed-off-by: Jeffrey Aaron Jeyasingh <jeffreyaaron06@gmail.com>

san81 mentioned this pull request Jun 23, 2025

Decoupling PipelineDataFlowModel dependency from PipelineTransformer #5809

Merged

4 tasks

dlvenable mentioned this pull request Jul 16, 2025

Consolidate CompressionOption enums #5876

Open

Uh oh!

Conversation

jeffreyAaron commented Jun 13, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Issues Resolved

Check List

Uh oh!

san81 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kkondaka left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

san81 left a comment

Choose a reason for hiding this comment

Uh oh!

dlvenable commented Jun 19, 2025

Uh oh!

dlvenable left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jeffreyAaron commented Jun 20, 2025

Uh oh!

jeffreyAaron commented Jun 20, 2025

Uh oh!

dlvenable left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

jeffreyAaron commented Jun 13, 2025 •

edited

Loading