Allow kafka streams configuration with external parameter

src/main/java/com/bakdata/common_kafka_streams/EnvironmentArgumentsParser.java

@@ -24,10 +24,12 @@
 
 package com.bakdata.common_kafka_streams;
 
-import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.StringJoiner;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 
 /**
@@ -64,15 +66,10 @@ public EnvironmentArgumentsParser(final String environmentPrefix, final String c
     }
 
     public List<String> parseVariables(final Map<String, String> environment) {
-        final List<String> environmentArguments = new ArrayList<>();
-        environment.forEach((k, v) -> {
-            if (!k.startsWith(this.environmentPrefix)) {
-                return;
-            }
-            environmentArguments.add(this.convertEnvironmentKeyToCommandLineParameter(k));
-            environmentArguments.add(v);
-        });
-        return environmentArguments;
+        return environment.entrySet().stream()
+                .filter(e -> e.getKey().startsWith(this.environmentPrefix))
+                .flatMap(this::convertEnvironmentVariable)
+                .collect(Collectors.toList());
     }
 
 
@@ -84,4 +81,10 @@ private String convertEnvironmentKeyToCommandLineParameter(final String environm
         }
         return "--" + sj;
     }
+
+    private Stream<String> convertEnvironmentVariable(final Entry<String, String> environmentEntry) {
+        final String key = this.convertEnvironmentKeyToCommandLineParameter(environmentEntry.getKey());
+        return Stream.of(key, environmentEntry.getValue());
+    }
+
 }

src/main/java/com/bakdata/common_kafka_streams/KafkaStreamsApplication.java

@@ -28,7 +28,9 @@
 import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
 import kafka.tools.StreamsResetter;
@@ -79,6 +81,9 @@ public abstract class KafkaStreamsApplication implements Runnable, AutoCloseable
                     + "might cause inconsistent processing with multiple replicas.")
     private boolean forceReprocessing = false;
 
+    @CommandLine.Option(names = "--streams-config", split = ",", description = "Additional Kafka Streams properties")
+    private Map<String, String> streamsConfig = new HashMap<>();
+
     @CommandLine.Option(names = "--input-topic", description = "Input topic")
     protected String inputTopic = "";
 
@@ -198,6 +203,8 @@ public Properties getKafkaProperties() {
         kafkaConfig.setProperty(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, this.getSchemaRegistryUrl());
         kafkaConfig.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, this.getBrokers());
 
+        this.streamsConfig.forEach(kafkaConfig::setProperty);
+
         return kafkaConfig;
     }
 

src/test/java/com/bakdata/common_kafka_streams/EnvironmentArgumentsParserTest.java

@@ -76,4 +76,11 @@ void shouldReturnEmptyArrayIfNoValidKeysPresent() {
         assertThat(result).isEmpty();
     }
 
+    @Test
+    void shouldConvertMapFormat() {
+        final List<String> result = this.parser.parseVariables(Map.of(
+                "STREAMS_STREAM_CONFIG", "consumer.acks=all,producer.acks=none"));
+        assertThat(result).containsExactly("--stream-config", "consumer.acks=all,producer.acks=none");
+    }
+
 }

src/test/java/com/bakdata/common_kafka_streams/WordCountTest.java

@@ -24,6 +24,8 @@
 
 package com.bakdata.common_kafka_streams;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 import com.bakdata.common_kafka_streams.test_applications.WordCount;
 import com.bakdata.fluent_kafka_streams_tests.junit5.TestTopologyExtension;
 import org.apache.kafka.common.serialization.Serdes;
@@ -33,7 +35,8 @@
 
 class WordCountTest {
     private static final String[] ARGS = {"--input-topic", "Input", "--output-topic", "Output",
-            "--brokers", "localhost:9092", "--schema-registry-url", "registryUrl"};
+            "--brokers", "localhost:9092", "--schema-registry-url", "registryUrl", "--streams-config",
+            "test.ack=1,test1.ack=2"};
     private final WordCount app = CommandLine.populateCommand(new WordCount(), ARGS);
 
     @RegisterExtension
@@ -52,4 +55,11 @@ void shouldAggregateSameWordStream() {
                 .expectNextRecord().hasKey("bla").hasValue(2L)
                 .expectNoMoreRecord();
     }
+
+    @Test
+    void shouldSetKafkaProperties() {
+        assertThat(this.app.getKafkaProperties().getProperty("test.ack")).isEqualTo("1");
+        assertThat(this.app.getKafkaProperties().getProperty("test1.ack")).isEqualTo("2");
+    }
+
 }

src/test/java/com/bakdata/common_kafka_streams/integration/ReprocessingTest.java

@@ -32,6 +32,7 @@
 import com.bakdata.schemaregistrymock.junit5.SchemaRegistryMockExtension;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 import net.mguenther.kafka.junit.EmbeddedKafkaCluster;
 import net.mguenther.kafka.junit.KeyValue;
 import net.mguenther.kafka.junit.ReadKeyValues;
@@ -64,6 +65,9 @@ void setup() {
         this.mirror.setOutputTopic(outputTopicName);
         this.mirror.setBrokers(this.kafkaCluster.getBrokerList());
         this.mirror.setProductive(false);
+        this.mirror.setStreamsConfig(
+                Map.of("default.value.serde", "org.apache.kafka.common.serialization.Serdes$StringSerde",
+                        "default.key.serde", "org.apache.kafka.common.serialization.Serdes$StringSerde"));
 
         this.kafkaCluster.createTopic(TopicConfig.forTopic(this.mirror.getOutputTopic()).useDefaults());
         this.kafkaCluster.createTopic(TopicConfig.forTopic(this.mirror.getInputTopic()).useDefaults());

src/test/java/com/bakdata/common_kafka_streams/test_applications/Mirror.java

@@ -25,11 +25,8 @@
 package com.bakdata.common_kafka_streams.test_applications;
 
 import com.bakdata.common_kafka_streams.KafkaStreamsApplication;
-import java.util.Properties;
 import lombok.NoArgsConstructor;
-import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.streams.StreamsBuilder;
-import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.kstream.KStream;
 
 @NoArgsConstructor
@@ -44,11 +41,4 @@ public void buildTopology(final StreamsBuilder builder) {
     public String getUniqueAppId() {
         return this.getClass().getSimpleName() + "-" + this.getInputTopic() + "-" + this.getOutputTopic();
     }
-
-    public Properties getKafkaProperties() {
-        final Properties kafkaConfig = super.getKafkaProperties();
-        kafkaConfig.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
-        kafkaConfig.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
-        return kafkaConfig;
-    }
 }