bakdata · philipp94831 · Mar 19, 2025 · Mar 19, 2025 · Mar 19, 2025 · Mar 19, 2025
diff --git a/build.gradle.kts b/build.gradle.kts
@@ -11,7 +11,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")
     }
 }
 

diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
@@ -4,7 +4,7 @@ mockito = "5.17.0"
 testcontainers = "1.21.0"
 
 [libraries]
-kafka-streams-utils = { group = "com.bakdata.kafka", name = "kafka-streams-utils", version = "1.1.1" }
+kafka-streams-utils = { group = "com.bakdata.kafka", name = "kafka-streams-utils", version = "1.2.0" }
 kafka-tools = { group = "org.apache.kafka", name = "kafka-tools" }
 kafka-streams = { group = "org.apache.kafka", name = "kafka-streams" }
 kafka-clients = { group = "org.apache.kafka", name = "kafka-clients" }

diff --git a/streams-bootstrap-cli-test/src/test/resources/log4j2.xml b/streams-bootstrap-cli-test/src/test/resources/log4j2.xml
@@ -27,6 +27,15 @@
         <Logger additivity="false" level="warn" name="org.apache.kafka.clients.admin.AdminClientConfig">
             <AppenderRef ref="Console"/>
         </Logger>
+        <Logger additivity="false" level="warn" name="org.apache.kafka.common.config.AbstractConfig">
+            <AppenderRef ref="Console"/>
+        </Logger>
+        <Logger additivity="false" level="warn" name="org.apache.kafka.clients.admin.internals.AdminMetadataManager">
+            <AppenderRef ref="Console"/>
+        </Logger>
+        <Logger additivity="false" level="warn" name="org.apache.kafka.clients.Metadata">
+            <AppenderRef ref="Console"/>
+        </Logger>
         <Root level="info">
             <AppenderRef ref="Console"/>
         </Root>

diff --git a/streams-bootstrap-cli/src/test/resources/log4j2.xml b/streams-bootstrap-cli/src/test/resources/log4j2.xml
@@ -30,5 +30,14 @@
         <Logger additivity="false" level="warn" name="org.apache.kafka.clients.admin.AdminClientConfig">
             <AppenderRef ref="Console"/>
         </Logger>
+        <Logger additivity="false" level="warn" name="org.apache.kafka.common.config.AbstractConfig">
+            <AppenderRef ref="Console"/>
+        </Logger>
+        <Logger additivity="false" level="warn" name="org.apache.kafka.clients.admin.internals.AdminMetadataManager">
+            <AppenderRef ref="Console"/>
+        </Logger>
+        <Logger additivity="false" level="warn" name="org.apache.kafka.clients.Metadata">
+            <AppenderRef ref="Console"/>
+        </Logger>
     </Loggers>
 </Configuration>
diff --git a/streams-bootstrap-core/build.gradle.kts b/streams-bootstrap-core/build.gradle.kts
@@ -7,9 +7,7 @@ plugins {
 
 dependencies {
     api(libs.kafka.streams.utils)
-    implementation(libs.kafka.tools) {
-        exclude(group = "org.slf4j", module = "slf4j-reload4j")
-    }
+    implementation(libs.kafka.tools)
 
     api(libs.kafka.streams)
     api(libs.kafka.clients)

diff --git a/streams-bootstrap-core/src/main/java/com/bakdata/kafka/ConfiguredProducerApp.java b/streams-bootstrap-core/src/main/java/com/bakdata/kafka/ConfiguredProducerApp.java
@@ -32,6 +32,7 @@
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.record.CompressionType;
 
 /**
  * A {@link ProducerApp} with a corresponding {@link ProducerTopicConfig}
@@ -50,7 +51,7 @@ private static Map<String, Object> createBaseConfig() {
         kafkaConfig.put(ProducerConfig.ACKS_CONFIG, "all");
 
         // compression
-        kafkaConfig.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "gzip");
+        kafkaConfig.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, CompressionType.GZIP.toString());
 
         return kafkaConfig;
     }

diff --git a/streams-bootstrap-core/src/main/java/com/bakdata/kafka/ConfiguredStreamsApp.java b/streams-bootstrap-core/src/main/java/com/bakdata/kafka/ConfiguredStreamsApp.java
@@ -31,6 +31,7 @@
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.record.CompressionType;
 import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.Topology;
 
@@ -54,7 +55,8 @@ private static Map<String, Object> createBaseConfig() {
         kafkaConfig.put(StreamsConfig.producerPrefix(ProducerConfig.ACKS_CONFIG), "all");
 
         // compression
-        kafkaConfig.put(StreamsConfig.producerPrefix(ProducerConfig.COMPRESSION_TYPE_CONFIG), "gzip");
+        kafkaConfig.put(StreamsConfig.producerPrefix(ProducerConfig.COMPRESSION_TYPE_CONFIG),
+                CompressionType.GZIP.toString());
 
         return kafkaConfig;
     }

diff --git a/streams-bootstrap-core/src/main/java/com/bakdata/kafka/ConsumedX.java b/streams-bootstrap-core/src/main/java/com/bakdata/kafka/ConsumedX.java
@@ -26,7 +26,8 @@
 
 import java.util.function.Function;
 import org.apache.kafka.common.serialization.Serde;
-import org.apache.kafka.streams.Topology.AutoOffsetReset;
+import org.apache.kafka.streams.AutoOffsetReset;
+import org.apache.kafka.streams.Topology;
 import org.apache.kafka.streams.kstream.Consumed;
 import org.apache.kafka.streams.processor.TimestampExtractor;
 
@@ -109,6 +110,14 @@ public static <K, V> ConsumedX<K, V> with(final TimestampExtractor timestampExtr
         return new ConsumedX<>(configurator -> Consumed.with(timestampExtractor));
     }
 
+    /**
+     * @see Consumed#with(Topology.AutoOffsetReset)
+     */
+    @Deprecated(since = "5.0.0")
+    public static <K, V> ConsumedX<K, V> with(final Topology.AutoOffsetReset resetPolicy) {
+        return new ConsumedX<>(configurator -> Consumed.with(resetPolicy));
+    }
+
     /**
      * @see Consumed#with(AutoOffsetReset)
      */
@@ -159,6 +168,14 @@ public ConsumedX<K, V> withOffsetResetPolicy(final AutoOffsetReset offsetResetPo
         return this.modify(consumed -> consumed.withOffsetResetPolicy(offsetResetPolicy));
     }
 
+    /**
+     * @see Consumed#withOffsetResetPolicy(Topology.AutoOffsetReset)
+     */
+    @Deprecated(since = "5.0.0")
+    public ConsumedX<K, V> withOffsetResetPolicy(final Topology.AutoOffsetReset offsetResetPolicy) {
+        return this.modify(consumed -> consumed.withOffsetResetPolicy(offsetResetPolicy));
+    }
+
     /**
      * @see Consumed#withTimestampExtractor(TimestampExtractor)
      */

diff --git a/streams-bootstrap-core/src/main/java/com/bakdata/kafka/KStreamX.java b/streams-bootstrap-core/src/main/java/com/bakdata/kafka/KStreamX.java
@@ -40,13 +40,10 @@
 import org.apache.kafka.streams.kstream.Produced;
 import org.apache.kafka.streams.kstream.Repartitioned;
 import org.apache.kafka.streams.kstream.StreamJoined;
-import org.apache.kafka.streams.kstream.TransformerSupplier;
 import org.apache.kafka.streams.kstream.ValueJoiner;
 import org.apache.kafka.streams.kstream.ValueJoinerWithKey;
 import org.apache.kafka.streams.kstream.ValueMapper;
 import org.apache.kafka.streams.kstream.ValueMapperWithKey;
-import org.apache.kafka.streams.kstream.ValueTransformerSupplier;
-import org.apache.kafka.streams.kstream.ValueTransformerWithKeySupplier;
 import org.apache.kafka.streams.processor.TopicNameExtractor;
 import org.apache.kafka.streams.processor.api.FixedKeyProcessorSupplier;
 import org.apache.kafka.streams.processor.api.ProcessorSupplier;
@@ -479,14 +476,6 @@ <VR> KErrorStreamX<K, V, K, VR> flatMapValuesCapturingErrors(
     @Override
     KStreamX<K, V> peek(ForeachAction<? super K, ? super V> action, Named named);
 
-    @Deprecated
-    @Override
-    KStreamX<K, V>[] branch(Named named, Predicate<? super K, ? super V>... predicates);
-
-    @Deprecated
-    @Override
-    KStreamX<K, V>[] branch(Predicate<? super K, ? super V>... predicates);
-
     @Override
     BranchedKStreamX<K, V> split();
 
@@ -499,14 +488,6 @@ <VR> KErrorStreamX<K, V, K, VR> flatMapValuesCapturingErrors(
     @Override
     KStreamX<K, V> merge(KStream<K, V> stream, Named named);
 
-    @Deprecated
-    @Override
-    KStreamX<K, V> through(String topic);
-
-    @Deprecated
-    @Override
-    KStreamX<K, V> through(String topic, Produced<K, V> produced);
-
     @Override
     KStreamX<K, V> repartition();
 
@@ -833,78 +814,6 @@ <GK, GV, RV> KStreamX<K, RV> leftJoin(GlobalKTable<GK, GV> globalTable,
             KeyValueMapper<? super K, ? super V, ? extends GK> keySelector,
             ValueJoinerWithKey<? super K, ? super V, ? super GV, ? extends RV> valueJoiner, Named named);
 
-    @Deprecated
-    @Override
-    <K1, V1> KStreamX<K1, V1> transform(
-            TransformerSupplier<? super K, ? super V, KeyValue<K1, V1>> transformerSupplier,
-            String... stateStoreNames);
-
-    @Deprecated
-    @Override
-    <K1, V1> KStreamX<K1, V1> transform(
-            TransformerSupplier<? super K, ? super V, KeyValue<K1, V1>> transformerSupplier,
-            Named named, String... stateStoreNames);
-
-    @Deprecated
-    @Override
-    <K1, V1> KStreamX<K1, V1> flatTransform(
-            TransformerSupplier<? super K, ? super V, Iterable<KeyValue<K1, V1>>> transformerSupplier,
-            String... stateStoreNames);
-
-    @Deprecated
-    @Override
-    <K1, V1> KStreamX<K1, V1> flatTransform(
-            TransformerSupplier<? super K, ? super V, Iterable<KeyValue<K1, V1>>> transformerSupplier, Named named,
-            String... stateStoreNames);
-
-    @Deprecated
-    @Override
-    <VR> KStreamX<K, VR> transformValues(
-            ValueTransformerSupplier<? super V, ? extends VR> valueTransformerSupplier,
-            String... stateStoreNames);
-
-    @Deprecated
-    @Override
-    <VR> KStreamX<K, VR> transformValues(
-            ValueTransformerSupplier<? super V, ? extends VR> valueTransformerSupplier,
-            Named named, String... stateStoreNames);
-
-    @Deprecated
-    @Override
-    <VR> KStreamX<K, VR> transformValues(
-            ValueTransformerWithKeySupplier<? super K, ? super V, ? extends VR> valueTransformerSupplier,
-            String... stateStoreNames);
-
-    @Deprecated
-    @Override
-    <VR> KStreamX<K, VR> transformValues(
-            ValueTransformerWithKeySupplier<? super K, ? super V, ? extends VR> valueTransformerSupplier, Named named,
-            String... stateStoreNames);
-
-    @Deprecated
-    @Override
-    <VR> KStreamX<K, VR> flatTransformValues(
-            ValueTransformerSupplier<? super V, Iterable<VR>> valueTransformerSupplier,
-            String... stateStoreNames);
-
-    @Deprecated
-    @Override
-    <VR> KStreamX<K, VR> flatTransformValues(
-            ValueTransformerSupplier<? super V, Iterable<VR>> valueTransformerSupplier,
-            Named named, String... stateStoreNames);
-
-    @Deprecated
-    @Override
-    <VR> KStreamX<K, VR> flatTransformValues(
-            ValueTransformerWithKeySupplier<? super K, ? super V, Iterable<VR>> valueTransformerSupplier,
-            String... stateStoreNames);
-
-    @Deprecated
-    @Override
-    <VR> KStreamX<K, VR> flatTransformValues(
-            ValueTransformerWithKeySupplier<? super K, ? super V, Iterable<VR>> valueTransformerSupplier, Named named,
-            String... stateStoreNames);
-
     @Override
     <KOut, VOut> KStreamX<KOut, VOut> process(
             ProcessorSupplier<? super K, ? super V, KOut, VOut> processorSupplier,