Upgrade to Kafka 3.8 (#267)

Author: philipp94831

gradle.properties

@@ -2,12 +2,12 @@ version=3.2.1-SNAPSHOT
 org.gradle.caching=true
 # running Kafka Streams in parallel causes problems with colliding consumer groups
 org.gradle.parallel=false
-kafkaVersion=3.7.1
+kafkaVersion=3.8.1
 testContainersVersion=1.20.4
-confluentVersion=7.7.0
-fluentKafkaVersion=2.15.0
-junitVersion=5.10.2
-mockitoVersion=5.11.0
-assertJVersion=3.25.3
-log4jVersion=2.23.1
+confluentVersion=7.8.0
+fluentKafkaVersion=2.16.0
+junitVersion=5.11.4
+mockitoVersion=5.15.2
+assertJVersion=3.27.2
+log4jVersion=2.24.3
 org.gradle.jvmargs=-Xmx4096m

streams-bootstrap-cli/build.gradle.kts

@@ -6,7 +6,7 @@ plugins {
 
 dependencies {
     api(project(":streams-bootstrap-core"))
-    api(group = "info.picocli", name = "picocli", version = "4.7.5")
+    api(group = "info.picocli", name = "picocli", version = "4.7.6")
 
     val junitVersion: String by project
     testRuntimeOnly(group = "org.junit.jupiter", name = "junit-jupiter-engine", version = junitVersion)

streams-bootstrap-cli/src/test/java/com/bakdata/kafka/integration/StreamsCleanUpTest.java

@@ -188,7 +188,7 @@ private List<KeyValue<String, Long>> readOutputTopic(final String outputTopic) {
                 .with(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class)
                 .from(outputTopic, TIMEOUT);
         return records.stream()
-                .map(record -> new KeyValue<>(record.key(), record.value()))
+                .map(consumerRecord -> new KeyValue<>(consumerRecord.key(), consumerRecord.value()))
                 .collect(Collectors.toList());
     }
 
@@ -213,7 +213,7 @@ private KafkaStreamsApplication<?> createWordCountApplication() {
     private <T extends KafkaStreamsApplication<?>> T configure(final T application) {
         application.setBootstrapServers(this.kafkaCluster.getBootstrapServers());
         application.setKafkaConfig(Map.of(
-                StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, "0",
+                StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, "0",
                 ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "10000"
         ));
         return application;

streams-bootstrap-core/build.gradle.kts

@@ -13,18 +13,18 @@ dependencies {
     val confluentVersion: String by project
     implementation(group = "io.confluent", name = "kafka-schema-serializer", version = confluentVersion)
     api(group = "io.confluent", name = "kafka-schema-registry-client", version = confluentVersion)
-    api(
+    implementation(
         group = "org.slf4j",
         name = "slf4j-api",
-        version = "2.0.9"
-    ) // required because other dependencies use Slf4j 1.x which is not properly resolved if this library is used in test scope
-    implementation(group = "org.jooq", name = "jool", version = "0.9.14")
+        version = "2.0.16"
+    )
+    implementation(group = "org.jooq", name = "jool", version = "0.9.15")
 
     val junitVersion: String by project
     testRuntimeOnly(group = "org.junit.jupiter", name = "junit-jupiter-engine", version = junitVersion)
     testImplementation(group = "org.junit.jupiter", name = "junit-jupiter-api", version = junitVersion)
     testImplementation(group = "org.junit.jupiter", name = "junit-jupiter-params", version = junitVersion)
-    testImplementation(group = "org.junit-pioneer", name = "junit-pioneer", version = "2.2.0")
+    testImplementation(group = "org.junit-pioneer", name = "junit-pioneer", version = "2.3.0")
     val assertJVersion: String by project
     testImplementation(group = "org.assertj", name = "assertj-core", version = assertJVersion)
     val mockitoVersion: String by project

streams-bootstrap-core/src/main/java/com/bakdata/kafka/util/ConsumerGroupClient.java

@@ -60,6 +60,14 @@ public static ConsumerGroupClient create(final Map<String, Object> configs, fina
         return new ConsumerGroupClient(AdminClient.create(configs), timeout);
     }
 
+    private static KafkaAdminException failedToDeleteGroup(final String groupName, final Throwable ex) {
+        return new KafkaAdminException("Failed to delete consumer group " + groupName, ex);
+    }
+
+    private static KafkaAdminException failedToListGroups(final Throwable ex) {
+        return new KafkaAdminException("Failed to list consumer groups", ex);
+    }
+
     /**
      * Delete a consumer group.
      *
@@ -74,14 +82,14 @@ public void deleteConsumerGroup(final String groupName) {
             log.info("Deleted consumer group '{}'", groupName);
         } catch (final InterruptedException ex) {
             Thread.currentThread().interrupt();
-            throw new KafkaAdminException("Failed to delete consumer group " + groupName, ex);
+            throw failedToDeleteGroup(groupName, ex);
         } catch (final ExecutionException ex) {
             if (ex.getCause() instanceof RuntimeException) {
                 throw (RuntimeException) ex.getCause();
             }
-            throw new KafkaAdminException("Failed to delete consumer group " + groupName, ex);
+            throw failedToDeleteGroup(groupName, ex);
         } catch (final TimeoutException ex) {
-            throw new KafkaAdminException("Failed to delete consumer group " + groupName, ex);
+            throw failedToDeleteGroup(groupName, ex);
         }
     }
 
@@ -115,14 +123,14 @@ public Collection<ConsumerGroupListing> listGroups() {
                     .get(this.timeout.toSeconds(), TimeUnit.SECONDS);
         } catch (final InterruptedException ex) {
             Thread.currentThread().interrupt();
-            throw new KafkaAdminException("Failed to list consumer groups", ex);
+            throw failedToListGroups(ex);
         } catch (final ExecutionException ex) {
             if (ex.getCause() instanceof RuntimeException) {
                 throw (RuntimeException) ex.getCause();
             }
-            throw new KafkaAdminException("Failed to list consumer groups", ex);
+            throw failedToListGroups(ex);
         } catch (final TimeoutException ex) {
-            throw new KafkaAdminException("Failed to list consumer groups", ex);
+            throw failedToListGroups(ex);
         }
     }
 

streams-bootstrap-core/src/test/java/com/bakdata/kafka/integration/ProducerCleanUpRunnerTest.java

@@ -176,7 +176,7 @@ private List<KeyValue<String, String>> readOutputTopic(final String outputTopic)
         final List<ConsumerRecord<String, String>> records =
                 this.newContainerHelper().read().from(outputTopic, Duration.ofSeconds(1L));
         return records.stream()
-                .map(record -> new org.apache.kafka.streams.KeyValue<>(record.key(), record.value()))
+                .map(StreamsCleanUpRunnerTest::toKeyValue)
                 .collect(Collectors.toList());
     }
 

streams-bootstrap-core/src/test/java/com/bakdata/kafka/integration/ProducerRunnerTest.java

@@ -74,7 +74,7 @@ private List<KeyValue<String, String>> readOutputTopic(final String outputTopic)
         final List<ConsumerRecord<String, String>> records =
                 this.newContainerHelper().read().from(outputTopic, Duration.ofSeconds(1L));
         return records.stream()
-                .map(record -> new KeyValue<>(record.key(), record.value()))
+                .map(StreamsCleanUpRunnerTest::toKeyValue)
                 .collect(Collectors.toList());
     }
 

streams-bootstrap-core/src/test/java/com/bakdata/kafka/integration/StreamsCleanUpRunnerTest.java

@@ -88,6 +88,10 @@ class StreamsCleanUpRunnerTest extends KafkaTest {
     @Mock
     private TopicHook topicHook;
 
+    static <K, V> KeyValue<K, V> toKeyValue(final ConsumerRecord<K, V> consumerRecord) {
+        return new KeyValue<>(consumerRecord.key(), consumerRecord.value());
+    }
+
     private static ConfiguredStreamsApp<StreamsApp> createWordCountPatternApplication() {
         return configureApp(new WordCountPattern(), StreamsTopicConfig.builder()
                 .inputPattern(Pattern.compile(".*_topic"))
@@ -547,7 +551,6 @@ void shouldDeleteSchemaOfInternalTopics()
         }
     }
 
-
     @Test
     void shouldDeleteSchemaOfIntermediateTopics()
             throws InterruptedException, IOException, RestClientException {
@@ -717,7 +720,7 @@ private List<KeyValue<String, Long>> readOutputTopic(final String outputTopic) {
                 .with(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class)
                 .from(outputTopic, TIMEOUT);
         return records.stream()
-                .map(record -> new KeyValue<>(record.key(), record.value()))
+                .map(StreamsCleanUpRunnerTest::toKeyValue)
                 .collect(Collectors.toList());
     }
 

streams-bootstrap-core/src/test/java/com/bakdata/kafka/integration/StreamsRunnerTest.java

@@ -94,7 +94,7 @@ static Thread run(final StreamsRunner runner) {
 
     static ConfiguredStreamsApp<StreamsApp> configureApp(final StreamsApp app, final StreamsTopicConfig topics) {
         final AppConfiguration<StreamsTopicConfig> configuration = new AppConfiguration<>(topics, Map.of(
-                StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, "0",
+                StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, "0",
                 ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "10000"
         ));
         return new ConfiguredStreamsApp<>(app, configuration);

streams-bootstrap-core/src/testFixtures/java/com/bakdata/kafka/TestUtil.java

@@ -31,6 +31,6 @@
 @UtilityClass
 public class TestUtil {
     public static KafkaContainer newKafkaCluster() {
-        return new KafkaContainer(DockerImageName.parse("apache/kafka:3.7.1"));
+        return new KafkaContainer(DockerImageName.parse("apache/kafka-native:3.8.1"));
     }
 }