Create v3

streams-bootstrap-picocli/src/main/java/com/bakdata/kafka/KafkaApplication.java

@@ -28,7 +28,6 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -149,21 +148,6 @@ protected KafkaEndpointConfig getEndpointConfig() {
                 .build();
     }
 
-    /**
-     * Kafka configurations overrides
-     * @return Kafka configuration
-     */
-    protected Map<String, Object> getKafkaConfigOverrides() {
-        return Collections.emptyMap();
-    }
-
-    protected Map<String, Object> getFullKafkaConfig() {
-        final Map<String, Object> config = new HashMap<>();
-        config.putAll(this.getKafkaConfigOverrides());
-        config.putAll(this.kafkaConfig);
-        return Collections.unmodifiableMap(config);
-    }
-
     private void startApplication() {
         log.info("Starting application");
         Runtime.getRuntime().addShutdownHook(new Thread(this::close));

streams-bootstrap-picocli/src/main/java/com/bakdata/kafka/KafkaProducerApplication.java

@@ -108,7 +108,7 @@ private ConfiguredProducerApp<ProducerApp> createConfiguredApp() {
 
     private ProducerAppConfiguration createConfiguration() {
         final ProducerTopicConfig topics = this.createTopicConfig();
-        final Map<String, Object> kafkaConfig = this.getFullKafkaConfig();
+        final Map<String, String> kafkaConfig = this.getKafkaConfig();
         return ProducerAppConfiguration.builder()
                 .topics(topics)
                 .kafkaConfig(kafkaConfig)

streams-bootstrap-picocli/src/main/java/com/bakdata/kafka/KafkaStreamsApplication.java

@@ -211,7 +211,7 @@ private ConfiguredStreamsApp<StreamsApp> createConfiguredApp(final boolean clean
 
     private StreamsAppConfiguration createConfiguration() {
         final StreamsTopicConfig topics = this.createTopicConfig();
-        final Map<String, Object> kafkaConfig = this.getFullKafkaConfig();
+        final Map<String, String> kafkaConfig = this.getKafkaConfig();
         final StreamsConfigurationOptions streamsOptions = this.createStreamsOptions();
         return StreamsAppConfiguration.builder()
                 .topics(topics)

streams-bootstrap/src/main/java/com/bakdata/kafka/ProducerApp.java

@@ -34,6 +34,10 @@
 @FunctionalInterface
 public interface ProducerApp extends AutoCloseable {
 
+    /**
+     * Setup Kafka resources, such as topics, before running this app
+     * @param configuration provides all runtime application configurations
+     */
     default void setup(final ProducerAppSetupConfiguration configuration) {
         // do nothing by default
     }

streams-bootstrap/src/main/java/com/bakdata/kafka/ProducerAppConfiguration.java

@@ -40,5 +40,5 @@ public class ProducerAppConfiguration {
     @Builder.Default
     @NonNull ProducerTopicConfig topics = ProducerTopicConfig.builder().build();
     @Builder.Default
-    @NonNull Map<String, Object> kafkaConfig = emptyMap();
+    @NonNull Map<String, ?> kafkaConfig = emptyMap();
 }

streams-bootstrap/src/main/java/com/bakdata/kafka/StreamsApp.java

@@ -35,6 +35,10 @@
 public interface StreamsApp extends AutoCloseable {
     int DEFAULT_PRODUCTIVE_REPLICATION_FACTOR = 3;
 
+    /**
+     * Setup Kafka resources, such as topics, before running this app
+     * @param configuration provides all runtime application configurations
+     */
     default void setup(final StreamsAppSetupConfiguration configuration) {
         // do nothing by default
     }

streams-bootstrap/src/main/java/com/bakdata/kafka/StreamsAppConfiguration.java

@@ -37,7 +37,7 @@ public class StreamsAppConfiguration {
     @Builder.Default
     @NonNull StreamsTopicConfig topics = StreamsTopicConfig.builder().build();
     @Builder.Default
-    @NonNull Map<String, Object> kafkaConfig = emptyMap();
+    @NonNull Map<String, ?> kafkaConfig = emptyMap();
     @Builder.Default
     @NonNull StreamsConfigurationOptions options = StreamsConfigurationOptions.builder().build();
 }