kafkaConfig = new HashMap<>();
-
- // exactly once and order
- kafkaConfig.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);
- kafkaConfig.put(ProducerConfig.ACKS_CONFIG, "all");
-
- // compression
- kafkaConfig.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "gzip");
-
- return kafkaConfig;
+ return emptyMap();
}
/**
diff --git a/streams-bootstrap/src/main/java/com/bakdata/kafka/StreamsApp.java b/streams-bootstrap/src/main/java/com/bakdata/kafka/StreamsApp.java
index f555bd5c..01f20552 100644
--- a/streams-bootstrap/src/main/java/com/bakdata/kafka/StreamsApp.java
+++ b/streams-bootstrap/src/main/java/com/bakdata/kafka/StreamsApp.java
@@ -24,16 +24,14 @@
package com.bakdata.kafka;
-import java.util.HashMap;
+import static java.util.Collections.emptyMap;
+
import java.util.Map;
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.streams.StreamsConfig;
/**
* Application that defines a Kafka Streams {@link org.apache.kafka.streams.Topology} and necessary configurations
*/
public interface StreamsApp extends AutoCloseable {
- int DEFAULT_PRODUCTIVE_REPLICATION_FACTOR = 3;
/**
* Setup Kafka resources, such as topics, before running this app
@@ -61,52 +59,11 @@ default void setup(final StreamsAppSetupConfiguration configuration) {
String getUniqueAppId(StreamsTopicConfig topics);
/**
- * This method should give a default configuration to run your streaming application with.
- * To add a custom configuration, add a similar method to your custom application class:
- * {@code
- * protected Map createKafkaProperties(StreamsOptions options) {
- * # Try to always use the kafka properties from the super class as base Map
- * Map kafkaConfig = StreamsApp.super.createKafkaProperties(options);
- * kafkaConfig.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, GenericAvroSerde.class);
- * kafkaConfig.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, GenericAvroSerde.class);
- * return kafkaConfig;
- * }
- * }
- *
- * Default configuration configures exactly-once, in-order, and compression:
- *
- * processing.guarantee=exactly_once_v2
- * producer.max.in.flight.requests.per.connection=1
- * producer.acks=all
- * producer.compression.type=gzip
- *
- *
- * If {@link StreamsConfigurationOptions#isProductive()} is set the following is configured additionally:
- *
- * replication.factor=3
- *
- *
- * @param options options to dynamically configure
- * @return Returns a default Kafka Streams configuration
+ * This method should give a default configuration to run your streaming application with.
+ * @return Returns a default Kafka Streams configuration. Empty by default
*/
- default Map createKafkaProperties(final StreamsConfigurationOptions options) {
- final Map kafkaConfig = new HashMap<>();
-
- // exactly once and order
- kafkaConfig.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
- kafkaConfig.put(StreamsConfig.producerPrefix(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION), 1);
-
- // resilience
- if (options.isProductive()) {
- kafkaConfig.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, DEFAULT_PRODUCTIVE_REPLICATION_FACTOR);
- }
-
- kafkaConfig.put(StreamsConfig.producerPrefix(ProducerConfig.ACKS_CONFIG), "all");
-
- // compression
- kafkaConfig.put(StreamsConfig.producerPrefix(ProducerConfig.COMPRESSION_TYPE_CONFIG), "gzip");
-
- return kafkaConfig;
+ default Map createKafkaProperties() {
+ return emptyMap();
}
/**
diff --git a/streams-bootstrap/src/main/java/com/bakdata/kafka/StreamsAppConfiguration.java b/streams-bootstrap/src/main/java/com/bakdata/kafka/StreamsAppConfiguration.java
index a6862c8c..b20f71c1 100644
--- a/streams-bootstrap/src/main/java/com/bakdata/kafka/StreamsAppConfiguration.java
+++ b/streams-bootstrap/src/main/java/com/bakdata/kafka/StreamsAppConfiguration.java
@@ -38,6 +38,4 @@ public class StreamsAppConfiguration {
@NonNull StreamsTopicConfig topics = StreamsTopicConfig.builder().build();
@Builder.Default
@NonNull Map kafkaConfig = emptyMap();
- @Builder.Default
- @NonNull StreamsConfigurationOptions options = StreamsConfigurationOptions.builder().build();
}
diff --git a/streams-bootstrap/src/main/java/com/bakdata/kafka/StreamsConfigurationOptions.java b/streams-bootstrap/src/main/java/com/bakdata/kafka/StreamsConfigurationOptions.java
deleted file mode 100644
index 9c32bc6c..00000000
--- a/streams-bootstrap/src/main/java/com/bakdata/kafka/StreamsConfigurationOptions.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * MIT License
- *
- * Copyright (c) 2024 bakdata
- *
- * Permission is hereby granted, free of charge, to any person obtaining a copy
- * of this software and associated documentation files (the "Software"), to deal
- * in the Software without restriction, including without limitation the rights
- * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
- * copies of the Software, and to permit persons to whom the Software is
- * furnished to do so, subject to the following conditions:
- *
- * The above copyright notice and this permission notice shall be included in all
- * copies or substantial portions of the Software.
- *
- * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
- * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
- * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
- * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
- * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
- * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
- * SOFTWARE.
- */
-
-package com.bakdata.kafka;
-
-import lombok.Builder;
-import lombok.Value;
-
-/**
- * Options to configure {@link StreamsApp#createKafkaProperties(StreamsConfigurationOptions)}
- */
-@Builder
-@Value
-public class StreamsConfigurationOptions {
-
- /**
- * Defines if {@code StreamsApp} should be configured for a productive environment and thus requires resilience
- */
- boolean productive;
-}
diff --git a/streams-bootstrap/src/test/java/com/bakdata/kafka/ConfiguredStreamsAppTest.java b/streams-bootstrap/src/test/java/com/bakdata/kafka/ConfiguredStreamsAppTest.java
index 82c6a8c9..97239e55 100644
--- a/streams-bootstrap/src/test/java/com/bakdata/kafka/ConfiguredStreamsAppTest.java
+++ b/streams-bootstrap/src/test/java/com/bakdata/kafka/ConfiguredStreamsAppTest.java
@@ -92,11 +92,11 @@ public String getUniqueAppId(final StreamsTopicConfig topics) {
}
@Override
- public Map createKafkaProperties(final StreamsConfigurationOptions options) {
- final Map properties = StreamsApp.super.createKafkaProperties(options);
- properties.put("foo", "bar");
- properties.put("hello", "world");
- return properties;
+ public Map createKafkaProperties() {
+ return Map.of(
+ "foo", "bar",
+ "hello", "world"
+ );
}
}
}
diff --git a/streams-bootstrap/src/test/java/com/bakdata/kafka/test_applications/ComplexTopologyApplication.java b/streams-bootstrap/src/test/java/com/bakdata/kafka/test_applications/ComplexTopologyApplication.java
index 97acdde5..d321e3ca 100644
--- a/streams-bootstrap/src/test/java/com/bakdata/kafka/test_applications/ComplexTopologyApplication.java
+++ b/streams-bootstrap/src/test/java/com/bakdata/kafka/test_applications/ComplexTopologyApplication.java
@@ -25,7 +25,6 @@
package com.bakdata.kafka.test_applications;
import com.bakdata.kafka.StreamsApp;
-import com.bakdata.kafka.StreamsConfigurationOptions;
import com.bakdata.kafka.StreamsTopicConfig;
import com.bakdata.kafka.TestRecord;
import com.bakdata.kafka.TopologyBuilder;
@@ -72,10 +71,10 @@ public String getUniqueAppId(final StreamsTopicConfig topics) {
}
@Override
- public Map createKafkaProperties(final StreamsConfigurationOptions options) {
- final Map kafkaConfig = StreamsApp.super.createKafkaProperties(options);
- kafkaConfig.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, StringSerde.class);
- kafkaConfig.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
- return kafkaConfig;
+ public Map createKafkaProperties() {
+ return Map.of(
+ StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, StringSerde.class,
+ StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class
+ );
}
}
diff --git a/streams-bootstrap/src/test/java/com/bakdata/kafka/test_applications/MirrorKeyWithAvro.java b/streams-bootstrap/src/test/java/com/bakdata/kafka/test_applications/MirrorKeyWithAvro.java
index 1e9a36e3..67056d93 100644
--- a/streams-bootstrap/src/test/java/com/bakdata/kafka/test_applications/MirrorKeyWithAvro.java
+++ b/streams-bootstrap/src/test/java/com/bakdata/kafka/test_applications/MirrorKeyWithAvro.java
@@ -25,7 +25,6 @@
package com.bakdata.kafka.test_applications;
import com.bakdata.kafka.StreamsApp;
-import com.bakdata.kafka.StreamsConfigurationOptions;
import com.bakdata.kafka.StreamsTopicConfig;
import com.bakdata.kafka.TestRecord;
import com.bakdata.kafka.TopologyBuilder;
@@ -50,10 +49,10 @@ public String getUniqueAppId(final StreamsTopicConfig topics) {
}
@Override
- public Map createKafkaProperties(final StreamsConfigurationOptions options) {
- final Map kafkaConfig = StreamsApp.super.createKafkaProperties(options);
- kafkaConfig.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
- kafkaConfig.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, StringSerde.class);
- return kafkaConfig;
+ public Map createKafkaProperties() {
+ return Map.of(
+ StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, SpecificAvroSerde.class,
+ StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, StringSerde.class
+ );
}
}
diff --git a/streams-bootstrap/src/test/java/com/bakdata/kafka/test_applications/MirrorValueWithAvro.java b/streams-bootstrap/src/test/java/com/bakdata/kafka/test_applications/MirrorValueWithAvro.java
index 26959f0f..a2ee1aa8 100644
--- a/streams-bootstrap/src/test/java/com/bakdata/kafka/test_applications/MirrorValueWithAvro.java
+++ b/streams-bootstrap/src/test/java/com/bakdata/kafka/test_applications/MirrorValueWithAvro.java
@@ -25,7 +25,6 @@
package com.bakdata.kafka.test_applications;
import com.bakdata.kafka.StreamsApp;
-import com.bakdata.kafka.StreamsConfigurationOptions;
import com.bakdata.kafka.StreamsTopicConfig;
import com.bakdata.kafka.TestRecord;
import com.bakdata.kafka.TopologyBuilder;
@@ -50,10 +49,10 @@ public String getUniqueAppId(final StreamsTopicConfig topics) {
}
@Override
- public Map createKafkaProperties(final StreamsConfigurationOptions options) {
- final Map kafkaConfig = StreamsApp.super.createKafkaProperties(options);
- kafkaConfig.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, StringSerde.class);
- kafkaConfig.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
- return kafkaConfig;
+ public Map createKafkaProperties() {
+ return Map.of(
+ StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, StringSerde.class,
+ StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class
+ );
}
}
diff --git a/streams-bootstrap/src/test/java/com/bakdata/kafka/test_applications/MirrorWithNonDefaultSerde.java b/streams-bootstrap/src/test/java/com/bakdata/kafka/test_applications/MirrorWithNonDefaultSerde.java
index 82606447..e947ab6f 100644
--- a/streams-bootstrap/src/test/java/com/bakdata/kafka/test_applications/MirrorWithNonDefaultSerde.java
+++ b/streams-bootstrap/src/test/java/com/bakdata/kafka/test_applications/MirrorWithNonDefaultSerde.java
@@ -25,7 +25,6 @@
package com.bakdata.kafka.test_applications;
import com.bakdata.kafka.StreamsApp;
-import com.bakdata.kafka.StreamsConfigurationOptions;
import com.bakdata.kafka.StreamsTopicConfig;
import com.bakdata.kafka.TestRecord;
import com.bakdata.kafka.TopologyBuilder;
@@ -41,6 +40,8 @@
@NoArgsConstructor
public class MirrorWithNonDefaultSerde implements StreamsApp {
+
+ boolean foo;
@Override
public void buildTopology(final TopologyBuilder builder) {
final Serde valueSerde = this.getValueSerde(builder.getKafkaProperties());
@@ -61,10 +62,10 @@ public String getUniqueAppId(final StreamsTopicConfig topics) {
}
@Override
- public Map createKafkaProperties(final StreamsConfigurationOptions options) {
- final Map kafkaConfig = StreamsApp.super.createKafkaProperties(options);
- kafkaConfig.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, StringSerde.class);
- kafkaConfig.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, StringSerde.class);
- return kafkaConfig;
+ public Map createKafkaProperties() {
+ return Map.of(
+ StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, StringSerde.class,
+ StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, StringSerde.class
+ );
}
}