From a7e090ea9dbf1f7a8ad41a5f53df41aea0f7086b Mon Sep 17 00:00:00 2001 From: Tom Billiet Date: Tue, 28 Jul 2026 13:59:17 +0200 Subject: [PATCH] Update dependecies, make it work with kafka4 clients --- README.md | 14 +++++++------- pom.xml | 16 +++++++--------- .../partitioners/DynamicStreamPartitioner.java | 4 ++-- src/test/java/PartitionerTest.java | 4 ++-- 4 files changed, 18 insertions(+), 20 deletions(-) diff --git a/README.md b/README.md index 843b5e0..9b0bb11 100644 --- a/README.md +++ b/README.md @@ -171,7 +171,7 @@ A very basic HTTP server is included in the SDK that can be used to return a hea Service myServer = new SimpleRestServer.Builder() .setListener("/health", () -> this.isHealthy()) - .setListener("/metrics", () -> this.scrapeMetrics()) + .setListener("/metrics", () -> this.scrapeMetrics()); myServer.start(); ``` @@ -180,7 +180,7 @@ A very basic HTTP server is included in the SDK that can be used to return a hea To initialize the SDK ```java -Sdk sdk = new Sdk.Builder().autoDetect().build() +Sdk sdk = new Sdk.Builder().autoDetect().build(); ``` > This will automatically detect if the SDK needs to be initialized from 'scratch' trough environment variables given by the DSH to the container, >or if it can be initialized from an existing configuration file that exists on the nodes @@ -190,11 +190,11 @@ Sdk sdk = new Sdk.Builder().autoDetect().build() from here on the `sdk` object can be used to initialize the parsers 1) to access Kafka specific configuration: ```java - KafkaConfigParser kp = KafkaConfigParser.of(sdk) + KafkaConfigParser kp = KafkaConfigParser.of(sdk); ``` 2) to access stream specific config: ```java - StreamsConfigParser sp = StreamsConfigParser.of(sdk) + StreamsConfigParser sp = StreamsConfigParser.of(sdk); ``` ### Standalone App @@ -205,15 +205,15 @@ The example blow describes a simple skeleton for an application on the DSH Sdk sdk = new Sdk.Builder().autoDetect().build(); KafkaConfigParser kafkaParser = KafkaConfigParser.of(sdk); StreamsConfigParser streamsParser = StreamsConfigParser.of(sdk); - KafkaClientFactory clientFactory = new KafkaClientFactory(streamsParser, kafkaParser); + KafkaClientFactory clientFactory = KafkaClientFactory.of(streamsParser, kafkaParser); // create a new kafka producer - KafkaProducer[KeyEnvelope, DataEnvelope] producer = clientFactory.createStreamProducer(null); + KafkaProducer producer = clientFactory.createStreamProducer(null); // create a new kafka consumer // that will be part of the same consumergroup as other instances of this application will be. - KafkaConsumer[KeyEnvelope, DataEnvelope] consumer = clientFactory.createSharedStreamConsumer(null); + KafkaConsumer consumer = clientFactory.createSharedStreamConsumer(null); List myStreams = ... // streams to consume the data from consumer.subscribe(subscriptionPatternFor(myStreams)); diff --git a/pom.xml b/pom.xml index 977e287..4e573a9 100755 --- a/pom.xml +++ b/pom.xml @@ -27,11 +27,11 @@ 1.6.2 - 3.21.12 - 2.0.1 + 4.35.1 + 2.0.18 1.4.12 - 1.78.1 - 3.2.3 + 1.84 + 4.3.1 3.8.1 2.22.2 @@ -40,10 +40,10 @@ 0.6.1 1.6.7 3.2.0 - 3.1.1 + 3.12.0 3.2.0 2.3 - 0.8.5 + 0.8.15 5.9.0 UTF-8 UTF-8 @@ -185,10 +185,8 @@ maven-compiler-plugin ${maven-compiler-plugin.version} - 1.8 - 1.8 + 11 utf-8 - 1.8 -Xlint:all true true diff --git a/src/main/java/dsh/sdk/kafka/partitioners/DynamicStreamPartitioner.java b/src/main/java/dsh/sdk/kafka/partitioners/DynamicStreamPartitioner.java index 61eae3c..cfc0828 100644 --- a/src/main/java/dsh/sdk/kafka/partitioners/DynamicStreamPartitioner.java +++ b/src/main/java/dsh/sdk/kafka/partitioners/DynamicStreamPartitioner.java @@ -3,7 +3,7 @@ import dsh.messages.Envelope; import dsh.sdk.streams.StreamsConfigParser; import org.apache.kafka.clients.producer.Partitioner; -import org.apache.kafka.clients.producer.internals.DefaultPartitioner; +import org.apache.kafka.clients.producer.RoundRobinPartitioner; import org.apache.kafka.common.Cluster; import java.util.Map; @@ -14,7 +14,7 @@ public class DynamicStreamPartitioner implements Partitioner { private DshStreamPartitioner underlying; - private final Partitioner defaultPartitioner = new DefaultPartitioner(); + private final Partitioner defaultPartitioner = new RoundRobinPartitioner(); public static final String CONFIG_KEY = "__dynamicstreampartitioner"; /** diff --git a/src/test/java/PartitionerTest.java b/src/test/java/PartitionerTest.java index 8fecc8a..5b4c608 100644 --- a/src/test/java/PartitionerTest.java +++ b/src/test/java/PartitionerTest.java @@ -3,7 +3,7 @@ import dsh.sdk.streams.StreamsConfigParser; import mocks.MockKafka; import org.apache.kafka.clients.producer.Partitioner; -import org.apache.kafka.clients.producer.internals.DefaultPartitioner; +import org.apache.kafka.clients.producer.RoundRobinPartitioner; import org.apache.kafka.common.Cluster; import org.apache.kafka.common.utils.Utils; import org.junit.jupiter.api.Tag; @@ -24,7 +24,7 @@ public class PartitionerTest { private Partitioner dynamicPartitioner = new DynamicStreamPartitioner(); - private Partitioner defaultPartitioner = new DefaultPartitioner(); + private Partitioner defaultPartitioner = new RoundRobinPartitioner(); private Cluster mockCluster = MockKafka.clusterFor("stream.my-topic", 24); @Test