Upgrade to Kafka 4.1 (#73)

Author: philipp94831

build.gradle.kts

@@ -16,7 +16,7 @@ allprojects {
     repositories {
         mavenCentral()
         maven(url = "https://packages.confluent.io/maven/")
-        maven(url = "https://s01.oss.sonatype.org/content/repositories/snapshots")
+        maven(url = "https://central.sonatype.com/repository/maven-snapshots")
     }
 }
 

error-handling-avro/build.gradle.kts

@@ -26,7 +26,9 @@ dependencies {
     testImplementation(libs.mockito.junit)
     testImplementation(libs.assertj)
     testImplementation(libs.log4j.slf4j2)
-    testImplementation(libs.kafka.streams.avro.serde)
+    testImplementation(libs.kafka.streams.avro.serde) {
+        exclude(group = "org.apache.kafka") // force usage of OSS kafka-clients
+    }
 }
 
 avro {

error-handling-core/src/main/java/com/bakdata/kafka/ErrorCapturingProcessor.java

@@ -68,7 +68,7 @@ public final class ErrorCapturingProcessor<K, V, KR, VR>
      * @see ErrorUtil#isRecoverable(Exception)
      */
     public static <K, V, KR, VR> Processor<K, V, KR, ProcessedKeyValue<K, V, VR>> captureErrors(
-            final @NonNull Processor<? super K, ? super V, KR, VR> processor) {
+            final @NonNull Processor<? super K, ? super V, ? extends KR, ? extends VR> processor) {
         return captureErrors(processor, ErrorUtil::isRecoverable);
     }
 
@@ -91,7 +91,7 @@ public static <K, V, KR, VR> Processor<K, V, KR, ProcessedKeyValue<K, V, VR>> ca
      * @return {@code Processor}
      */
     public static <K, V, KR, VR> Processor<K, V, KR, ProcessedKeyValue<K, V, VR>> captureErrors(
-            final @NonNull Processor<? super K, ? super V, KR, VR> processor,
+            final @NonNull Processor<? super K, ? super V, ? extends KR, ? extends VR> processor,
             final @NonNull Predicate<Exception> errorFilter) {
         return new ErrorCapturingProcessor<>((Processor<K, V, KR, VR>) processor, errorFilter);
     }
@@ -110,7 +110,7 @@ public static <K, V, KR, VR> Processor<K, V, KR, ProcessedKeyValue<K, V, VR>> ca
      * @see ErrorUtil#isRecoverable(Exception)
      */
     public static <K, V, KR, VR> ProcessorSupplier<K, V, KR, ProcessedKeyValue<K, V, VR>> captureErrors(
-            final @NonNull ProcessorSupplier<? super K, ? super V, KR, VR> supplier) {
+            final @NonNull ProcessorSupplier<? super K, ? super V, ? extends KR, ? extends VR> supplier) {
         return captureErrors(supplier, ErrorUtil::isRecoverable);
     }
 
@@ -134,7 +134,7 @@ public static <K, V, KR, VR> ProcessorSupplier<K, V, KR, ProcessedKeyValue<K, V,
      * @return {@code ProcessorSupplier}
      */
     public static <K, V, KR, VR> ProcessorSupplier<K, V, KR, ProcessedKeyValue<K, V, VR>> captureErrors(
-            final @NonNull ProcessorSupplier<? super K, ? super V, KR, VR> supplier,
+            final @NonNull ProcessorSupplier<? super K, ? super V, ? extends KR, ? extends VR> supplier,
             final @NonNull Predicate<Exception> errorFilter) {
         return new ProcessorSupplier<>() {
             @Override

error-handling-core/src/main/java/com/bakdata/kafka/ErrorCapturingValueProcessor.java

@@ -66,7 +66,7 @@ public final class ErrorCapturingValueProcessor<K, V, VR>
      * @see ErrorUtil#isRecoverable(Exception)
      */
     public static <K, V, VR> FixedKeyProcessor<K, V, ProcessedValue<V, VR>> captureErrors(
-            final @NonNull FixedKeyProcessor<? super K, ? super V, VR> processor) {
+            final @NonNull FixedKeyProcessor<? super K, ? super V, ? extends VR> processor) {
         return captureErrors(processor, ErrorUtil::isRecoverable);
     }
 
@@ -88,7 +88,7 @@ public static <K, V, VR> FixedKeyProcessor<K, V, ProcessedValue<V, VR>> captureE
      * @return {@code FixedKeyProcessor}
      */
     public static <K, V, VR> FixedKeyProcessor<K, V, ProcessedValue<V, VR>> captureErrors(
-            final @NonNull FixedKeyProcessor<? super K, ? super V, VR> processor,
+            final @NonNull FixedKeyProcessor<? super K, ? super V, ? extends VR> processor,
             final @NonNull Predicate<Exception> errorFilter) {
         return new ErrorCapturingValueProcessor<>((FixedKeyProcessor<K, V, VR>) processor, errorFilter);
     }
@@ -106,7 +106,7 @@ public static <K, V, VR> FixedKeyProcessor<K, V, ProcessedValue<V, VR>> captureE
      * @see ErrorUtil#isRecoverable(Exception)
      */
     public static <K, V, VR> FixedKeyProcessorSupplier<K, V, ProcessedValue<V, VR>> captureErrors(
-            final @NonNull FixedKeyProcessorSupplier<? super K, ? super V, VR> supplier) {
+            final @NonNull FixedKeyProcessorSupplier<? super K, ? super V, ? extends VR> supplier) {
         return captureErrors(supplier, ErrorUtil::isRecoverable);
     }
 
@@ -129,7 +129,7 @@ public static <K, V, VR> FixedKeyProcessorSupplier<K, V, ProcessedValue<V, VR>>
      * @return {@code FixedKeyProcessorSupplier}
      */
     public static <K, V, VR> FixedKeyProcessorSupplier<K, V, ProcessedValue<V, VR>> captureErrors(
-            final @NonNull FixedKeyProcessorSupplier<? super K, ? super V, VR> supplier,
+            final @NonNull FixedKeyProcessorSupplier<? super K, ? super V, ? extends VR> supplier,
             final @NonNull Predicate<Exception> errorFilter) {
         return new FixedKeyProcessorSupplier<>() {
             @Override

error-handling-proto/build.gradle.kts

@@ -17,7 +17,9 @@ dependencies {
     testImplementation(libs.mockito.junit)
     testImplementation(libs.assertj)
     testImplementation(libs.log4j.slf4j2)
-    testImplementation(libs.kafka.streams.protobuf.serde)
+    testImplementation(libs.kafka.streams.protobuf.serde) {
+        exclude(group = "org.apache.kafka") // force usage of OSS kafka-clients
+    }
 }
 
 protobuf {

gradle/libs.versions.toml

@@ -4,7 +4,7 @@ junit = "5.14.0"
 mockito = "5.20.0"
 
 [libraries]
-kafka-bom = { group = "com.bakdata.kafka", name = "kafka-bom", version = "1.2.1" }
+kafka-bom = { group = "com.bakdata.kafka", name = "kafka-bom", version = "1.3.0" }
 kafka-streams = { group = "org.apache.kafka", name = "kafka-streams" }
 kafka-streams-avro-serde = { group = "io.confluent", name = "kafka-streams-avro-serde" }
 kafka-streams-protobuf-serde = { group = "io.confluent", name = "kafka-streams-protobuf-serde" }
@@ -21,7 +21,7 @@ junit-jupiter = { group = "org.junit.jupiter", name = "junit-jupiter", version.r
 assertj = { group = "org.assertj", name = "assertj-core", version = "3.27.6" }
 mockito-core = { group = "org.mockito", name = "mockito-core", version.ref = "mockito" }
 mockito-junit = { group = "org.mockito", name = "mockito-junit-jupiter", version.ref = "mockito" }
-fluentKafkaStreamsTests = { group = "com.bakdata.fluent-kafka-streams-tests", name = "fluent-kafka-streams-tests-junit5", version = "3.4.1" }
+fluentKafkaStreamsTests = { group = "com.bakdata.fluent-kafka-streams-tests", name = "fluent-kafka-streams-tests-junit5", version = "3.5.0" }
 log4j-slf4j2 = { group = "org.apache.logging.log4j", name = "log4j-slf4j2-impl", version = "2.25.2" }
 
 [plugins]