Upgrade to Kafka 3.8 and Avro 1.12

bakdata · Jan 7, 2025 · 40318f6 · 40318f6
1 parent 5f2389d
commit 40318f6
Show file tree

Hide file tree

Showing 33 changed files with 139 additions and 143 deletions.
diff --git a/...ng-core/src/test/java/com/bakdata/kafka/ErrorCapturingFlatKeyValueMapperTopologyTest.java b/...ng-core/src/test/java/com/bakdata/kafka/ErrorCapturingFlatKeyValueMapperTopologyTest.java
@@ -100,7 +100,7 @@ void shouldForwardRecoverableException(final SoftAssertions softly) {
                         .add(1, "foo"))
                 .hasCause(throwable);
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -136,7 +136,7 @@ void shouldCaptureKeyValueMapperError(final SoftAssertions softly) {
                 .add(1, "foo")
                 .add(2, "bar");
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -179,7 +179,7 @@ void shouldHandleNullInput(final SoftAssertions softly) {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -204,7 +204,7 @@ void shouldHandleErrorOnNullInput(final SoftAssertions softly) {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -241,7 +241,7 @@ void shouldHandleNullKeyValue(final SoftAssertions softly) {
                 .withValueSerde(STRING_SERDE)
                 .add(2, "bar");
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)

diff --git a/...dling-core/src/test/java/com/bakdata/kafka/ErrorCapturingFlatTransformerTopologyTest.java b/...dling-core/src/test/java/com/bakdata/kafka/ErrorCapturingFlatTransformerTopologyTest.java
@@ -125,7 +125,7 @@ public void close() {
                         .add(1, "foo"))
                 .hasCause(throwable);
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -206,7 +206,7 @@ public void close() {
                 .add(2, "bar")
                 .add(3, "baz");
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -266,7 +266,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -312,7 +312,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -355,7 +355,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -410,7 +410,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(2, "bar");
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -456,7 +456,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(2, "bar");
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)

diff --git a/...-core/src/test/java/com/bakdata/kafka/ErrorCapturingFlatValueTransformerTopologyTest.java b/...-core/src/test/java/com/bakdata/kafka/ErrorCapturingFlatValueTransformerTopologyTest.java
@@ -193,7 +193,7 @@ public void close() {
                 .add(1, "foo")
                 .add(2, "bar");
         final List<ProducerRecord<Integer, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(INTEGER_SERDE)
+                .withKeySerde(INTEGER_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -251,7 +251,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Integer, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(INTEGER_SERDE)
+                .withKeySerde(INTEGER_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -293,7 +293,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Integer, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(INTEGER_SERDE)
+                .withKeySerde(INTEGER_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)

diff --git a/...rc/test/java/com/bakdata/kafka/ErrorCapturingFlatValueTransformerWithKeyTopologyTest.java b/...rc/test/java/com/bakdata/kafka/ErrorCapturingFlatValueTransformerWithKeyTopologyTest.java
@@ -196,7 +196,7 @@ public void close() {
                 .add(1, "foo")
                 .add(2, "bar");
         final List<ProducerRecord<Integer, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(INTEGER_SERDE)
+                .withKeySerde(INTEGER_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -254,7 +254,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Integer, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(INTEGER_SERDE)
+                .withKeySerde(INTEGER_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -296,7 +296,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Integer, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(INTEGER_SERDE)
+                .withKeySerde(INTEGER_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)

diff --git a/...ndling-core/src/test/java/com/bakdata/kafka/ErrorCapturingKeyValueMapperTopologyTest.java b/...ndling-core/src/test/java/com/bakdata/kafka/ErrorCapturingKeyValueMapperTopologyTest.java
@@ -100,7 +100,7 @@ void shouldForwardRecoverableException(final SoftAssertions softly) {
                         .add(1, "foo"))
                 .hasCause(throwable);
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -134,7 +134,7 @@ void shouldCaptureKeyValueMapperError(final SoftAssertions softly) {
                 .add(1, "foo")
                 .add(2, "bar");
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -176,7 +176,7 @@ void shouldHandleNullInput(final SoftAssertions softly) {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -202,7 +202,7 @@ void shouldHandleErrorOnNullInput(final SoftAssertions softly) {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -238,7 +238,7 @@ void shouldHandleNullKeyValue(final SoftAssertions softly) {
                 .withValueSerde(STRING_SERDE)
                 .add(2, "bar");
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)

diff --git a/error-handling-core/src/test/java/com/bakdata/kafka/ErrorCapturingProcessorTopologyTest.java b/error-handling-core/src/test/java/com/bakdata/kafka/ErrorCapturingProcessorTopologyTest.java
@@ -121,7 +121,7 @@ public void close() {
                         .add(1, "foo"))
                 .hasCause(throwable);
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -197,7 +197,7 @@ public void close() {
                 .add(1, "foo")
                 .add(2, "bar");
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -262,7 +262,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -308,7 +308,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -366,7 +366,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -420,7 +420,7 @@ public void close() {
                 .add(2, "bar")
                 .add(3, "baz");
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)

diff --git a/...-handling-core/src/test/java/com/bakdata/kafka/ErrorCapturingTransformerTopologyTest.java b/...-handling-core/src/test/java/com/bakdata/kafka/ErrorCapturingTransformerTopologyTest.java
@@ -123,7 +123,7 @@ public void close() {
                         .add(1, "foo"))
                 .hasCause(throwable);
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -203,7 +203,7 @@ public void close() {
                 .add(2, "bar")
                 .add(3, "baz");
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -272,7 +272,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -318,7 +318,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -375,7 +375,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -427,7 +427,7 @@ public void close() {
                 .add(2, "bar")
                 .add(3, "baz");
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -485,7 +485,7 @@ public void close() {
                 .add(2, "bar")
                 .add(3, "baz");
         final List<ProducerRecord<Double, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(DOUBLE_SERDE)
+                .withKeySerde(DOUBLE_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)

diff --git a/...ndling-core/src/test/java/com/bakdata/kafka/ErrorCapturingValueProcessorTopologyTest.java b/...ndling-core/src/test/java/com/bakdata/kafka/ErrorCapturingValueProcessorTopologyTest.java
@@ -195,7 +195,7 @@ public void close() {
                 .add(1, "foo")
                 .add(2, "bar");
         final List<ProducerRecord<Integer, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(INTEGER_SERDE)
+                .withKeySerde(INTEGER_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -260,7 +260,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Integer, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(INTEGER_SERDE)
+                .withKeySerde(INTEGER_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -306,7 +306,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Integer, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(INTEGER_SERDE)
+                .withKeySerde(INTEGER_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -364,7 +364,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(2, "bar");
         final List<ProducerRecord<Integer, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(INTEGER_SERDE)
+                .withKeySerde(INTEGER_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)

diff --git a/...ling-core/src/test/java/com/bakdata/kafka/ErrorCapturingValueTransformerTopologyTest.java b/...ling-core/src/test/java/com/bakdata/kafka/ErrorCapturingValueTransformerTopologyTest.java
@@ -190,7 +190,7 @@ public void close() {
                 .add(1, "foo")
                 .add(2, "bar");
         final List<ProducerRecord<Integer, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(INTEGER_SERDE)
+                .withKeySerde(INTEGER_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -252,7 +252,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Integer, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(INTEGER_SERDE)
+                .withKeySerde(INTEGER_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -298,7 +298,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(null, null);
         final List<ProducerRecord<Integer, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(INTEGER_SERDE)
+                .withKeySerde(INTEGER_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)
@@ -353,7 +353,7 @@ public void close() {
                 .withValueSerde(STRING_SERDE)
                 .add(2, "bar");
         final List<ProducerRecord<Integer, Long>> records = this.topology.streamOutput(OUTPUT_TOPIC)
-                        .withKeySerde(INTEGER_SERDE)
+                .withKeySerde(INTEGER_SERDE)
                 .withValueSerde(LONG_SERDE)
                 .toList();
         softly.assertThat(records)