Use Map instead of Properties

fluent-kafka-streams-tests-junit4/src/main/java/com/bakdata/fluent_kafka_streams_tests/junit4/TestTopologyRule.java

@@ -1,7 +1,7 @@
 /*
  * MIT License
  *
- * Copyright (c) 2023 bakdata
+ * 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
@@ -27,7 +27,6 @@
 import com.bakdata.fluent_kafka_streams_tests.TestTopology;
 import com.bakdata.schemaregistrymock.SchemaRegistryMock;
 import java.util.Map;
-import java.util.Properties;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import lombok.Getter;
@@ -72,39 +71,39 @@
 public class TestTopologyRule<DefaultK, DefaultV> extends TestTopology<DefaultK, DefaultV>
         implements TestRule {
     public TestTopologyRule(
-            final Function<? super Properties, ? extends Topology> topologyFactory,
-            final Function<? super String, ? extends Map<?, ?>> propertiesFactory) {
+            final Function<? super Map<String, Object>, ? extends Topology> topologyFactory,
+            final Function<? super String, ? extends Map<String, ?>> propertiesFactory) {
         super(topologyFactory, propertiesFactory);
     }
     public TestTopologyRule(
-            final Function<? super Properties, ? extends Topology> topologyFactory,
-            final Map<Object, Object> properties) {
+            final Function<? super Map<String, Object>, ? extends Topology> topologyFactory,
+            final Map<String, Object> properties) {
         super(topologyFactory, properties);
     }
 
     public TestTopologyRule(
             final Supplier<? extends Topology> topologyFactory,
-            final Function<? super String, ? extends Map<?, ?>> propertiesFactory) {
+            final Function<? super String, ? extends Map<String, ?>> propertiesFactory) {
         super(topologyFactory, propertiesFactory);
     }
 
     public TestTopologyRule(
-            final Supplier<? extends Topology> topologyFactory, final Map<Object, Object> properties) {
+            final Supplier<? extends Topology> topologyFactory, final Map<String, Object> properties) {
         super(topologyFactory, properties);
     }
 
     public TestTopologyRule(final Topology topology,
-            final Function<? super String, ? extends Map<?, ?>> propertiesFactory) {
+            final Function<? super String, ? extends Map<String, ?>> propertiesFactory) {
         super(topology, propertiesFactory);
     }
 
-    public TestTopologyRule(final Topology topology, final Map<Object, Object> properties) {
+    public TestTopologyRule(final Topology topology, final Map<String, Object> properties) {
         super(topology, properties);
     }
 
     protected TestTopologyRule(
-            final Function<? super Properties, ? extends Topology> topologyFactory,
-            final Function<? super String, ? extends Map<?, ?>> propertiesFactory,
+            final Function<? super Map<String, Object>, ? extends Topology> topologyFactory,
+            final Function<? super String, ? extends Map<String, ?>> propertiesFactory,
             final Serde<DefaultK> defaultKeySerde, final Serde<DefaultV> defaultValueSerde,
             final SchemaRegistryMock schemaRegistryMock) {
         super(topologyFactory, propertiesFactory, defaultKeySerde, defaultValueSerde, schemaRegistryMock);
@@ -126,8 +125,9 @@ public void evaluate() throws Throwable {
     }
 
     @Override
-    protected <K, V> TestTopologyRule<K, V> with(final Function<? super Properties, ? extends Topology> topologyFactory,
-            final Function<? super String, ? extends Map<?, ?>> propertiesFactory, final Serde<K> defaultKeySerde,
+    protected <K, V> TestTopologyRule<K, V> with(
+            final Function<? super Map<String, Object>, ? extends Topology> topologyFactory,
+            final Function<? super String, ? extends Map<String, ?>> propertiesFactory, final Serde<K> defaultKeySerde,
             final Serde<V> defaultValueSerde,
             final SchemaRegistryMock schemaRegistryMock) {
         return new TestTopologyRule<>(topologyFactory, propertiesFactory, defaultKeySerde, defaultValueSerde,

fluent-kafka-streams-tests-junit4/src/test/java/com/bakdata/fluent_kafka_streams_tests/junit4/WordCountTest.java

@@ -41,7 +41,7 @@ public class WordCountTest {
 
     @Rule
     public final TestTopologyRule<Object, String> testTopology =
-            new TestTopologyRule<>(this.app::getTopology, this.app.getKafkaProperties());
+            new TestTopologyRule<>(this.app::getTopology, WordCount.getKafkaProperties());
 
     @Test
     public void shouldAggregateSameWordStream() {

fluent-kafka-streams-tests-junit4/src/test/java/com/bakdata/fluent_kafka_streams_tests/junit4/WordCountWithStaticTopologyTest.java

@@ -1,7 +1,7 @@
 /*
  * MIT License
  *
- * Copyright (c) 2019 bakdata GmbH
+ * 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
@@ -34,7 +34,7 @@ public class WordCountWithStaticTopologyTest {
 
     @Rule
     public final TestTopologyRule<Object, String> testTopology = new TestTopologyRule<>(this.app.getTopology(),
-            this.app.getKafkaProperties());
+            WordCount.getKafkaProperties());
 
     @Test
     public void shouldAggregateSameWordStream() {

fluent-kafka-streams-tests-junit4/src/test/java/com/bakdata/fluent_kafka_streams_tests/junit4/WordCountWitherTest.java

@@ -1,7 +1,7 @@
 /*
  * MIT License
  *
- * Copyright (c) 2019 bakdata GmbH
+ * 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
@@ -37,7 +37,7 @@ public class WordCountWitherTest {
 
     @Rule
     public final TestTopologyRule<Object, String> testTopology =
-            new TestTopologyRule<>(this.app.getTopology(), this.app.getKafkaProperties())
+            new TestTopologyRule<>(this.app.getTopology(), WordCount.getKafkaProperties())
                     .withDefaultValueSerde(Serdes.String())
                     .withSchemaRegistryMock(new SchemaRegistryMock(List.of(new AvroSchemaProvider())));
 

fluent-kafka-streams-tests-junit4/src/test/java/com/bakdata/fluent_kafka_streams_tests/junit4/test_applications/WordCount.java

@@ -1,40 +1,57 @@
+/*
+ * 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.fluent_kafka_streams_tests.junit4.test_applications;
 
 import java.util.Arrays;
-import java.util.Properties;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.regex.Pattern;
 import lombok.Getter;
 import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.common.serialization.Serdes;
-import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.common.serialization.Serdes.StringSerde;
 import org.apache.kafka.streams.StreamsBuilder;
 import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.Topology;
 import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.kstream.KTable;
 import org.apache.kafka.streams.kstream.Produced;
 
+@Getter
 public class WordCount {
-    @Getter
     private final String inputTopic = "wordcount-input";
 
-    @Getter
     private final String outputTopic = "wordcount-output";
 
-    public static void main(final String[] args) {
-        final WordCount wordCount = new WordCount();
-        final KafkaStreams streams = new KafkaStreams(wordCount.getTopology(), wordCount.getKafkaProperties());
-        streams.start();
-        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
-    }
-
-    public Properties getKafkaProperties() {
+    public static Map<String, Object> getKafkaProperties() {
         final String brokers = "localhost:9092";
-        final Properties kafkaConfig = new Properties();
-        kafkaConfig.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount");
-        kafkaConfig.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
-        kafkaConfig.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
-        kafkaConfig.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
+        final Map<String, Object> kafkaConfig = new HashMap<>();
+        kafkaConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount");
+        kafkaConfig.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
+        kafkaConfig.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, StringSerde.class);
+        kafkaConfig.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, StringSerde.class);
         return kafkaConfig;
     }
 

fluent-kafka-streams-tests-junit5/src/main/java/com/bakdata/fluent_kafka_streams_tests/junit5/TestTopologyExtension.java

@@ -1,7 +1,7 @@
 /*
  * MIT License
  *
- * Copyright (c) 2023 bakdata
+ * 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
@@ -27,7 +27,6 @@
 import com.bakdata.fluent_kafka_streams_tests.TestTopology;
 import com.bakdata.schemaregistrymock.SchemaRegistryMock;
 import java.util.Map;
-import java.util.Properties;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import lombok.Getter;
@@ -72,40 +71,40 @@
 public class TestTopologyExtension<DefaultK, DefaultV> extends TestTopology<DefaultK, DefaultV>
         implements BeforeEachCallback, AfterEachCallback {
     public TestTopologyExtension(
-            final Function<? super Properties, ? extends Topology> topologyFactory,
-            final Function<? super String, ? extends Map<?, ?>> propertiesFactory) {
+            final Function<? super Map<String, Object>, ? extends Topology> topologyFactory,
+            final Function<? super String, ? extends Map<String, ?>> propertiesFactory) {
         super(topologyFactory, propertiesFactory);
     }
 
     public TestTopologyExtension(
-            final Function<? super Properties, ? extends Topology> topologyFactory,
-            final Map<Object, Object> properties) {
+            final Function<? super Map<String, Object>, ? extends Topology> topologyFactory,
+            final Map<String, Object> properties) {
         super(topologyFactory, properties);
     }
 
     public TestTopologyExtension(
             final Supplier<? extends Topology> topologyFactory,
-            final Function<? super String, ? extends Map<?, ?>> propertiesFactory) {
+            final Function<? super String, ? extends Map<String, ?>> propertiesFactory) {
         super(topologyFactory, propertiesFactory);
     }
 
     public TestTopologyExtension(
-            final Supplier<? extends Topology> topologyFactory, final Map<Object, Object> properties) {
+            final Supplier<? extends Topology> topologyFactory, final Map<String, Object> properties) {
         super(topologyFactory, properties);
     }
 
     public TestTopologyExtension(final Topology topology,
-            final Function<? super String, ? extends Map<?, ?>> propertiesFactory) {
+            final Function<? super String, ? extends Map<String, ?>> propertiesFactory) {
         super(topology, propertiesFactory);
     }
 
-    public TestTopologyExtension(final Topology topology, final Map<Object, Object> properties) {
+    public TestTopologyExtension(final Topology topology, final Map<String, Object> properties) {
         super(topology, properties);
     }
 
     protected TestTopologyExtension(
-            final Function<? super Properties, ? extends Topology> topologyFactory,
-            final Function<? super String, ? extends Map<?, ?>> propertiesFactory,
+            final Function<? super Map<String, Object>, ? extends Topology> topologyFactory,
+            final Function<? super String, ? extends Map<String, ?>> propertiesFactory,
             final Serde<DefaultK> defaultKeySerde, final Serde<DefaultV> defaultValueSerde,
             final SchemaRegistryMock schemaRegistryMock) {
         super(topologyFactory, propertiesFactory, defaultKeySerde, defaultValueSerde, schemaRegistryMock);
@@ -122,8 +121,9 @@ public void beforeEach(final ExtensionContext context) {
     }
 
     @Override
-    protected <K, V> TestTopology<K, V> with(final Function<? super Properties, ? extends Topology> topologyFactory,
-            final Function<? super String, ? extends Map<?, ?>> propertiesFactory, final Serde<K> defaultKeySerde,
+    protected <K, V> TestTopology<K, V> with(
+            final Function<? super Map<String, Object>, ? extends Topology> topologyFactory,
+            final Function<? super String, ? extends Map<String, ?>> propertiesFactory, final Serde<K> defaultKeySerde,
             final Serde<V> defaultValueSerde,
             final SchemaRegistryMock schemaRegistry) {
         return new TestTopologyExtension<>(topologyFactory, propertiesFactory, defaultKeySerde, defaultValueSerde,

fluent-kafka-streams-tests-junit5/src/test/java/com/bakdata/fluent_kafka_streams_tests/junit5/WordCountTest.java

@@ -41,7 +41,7 @@ class WordCountTest {
 
     @RegisterExtension
     final TestTopologyExtension<Object, String> testTopology = new TestTopologyExtension<>(this.app::getTopology,
-            this.app.getKafkaProperties());
+            WordCount.getKafkaProperties());
 
     @Test
     void shouldAggregateSameWordStream() {

fluent-kafka-streams-tests-junit5/src/test/java/com/bakdata/fluent_kafka_streams_tests/junit5/WordCountWithStaticTopologyTest.java

@@ -1,7 +1,7 @@
 /*
  * MIT License
  *
- * Copyright (c) 2019 bakdata GmbH
+ * 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
@@ -34,7 +34,7 @@ class WordCountWithStaticTopologyTest {
 
     @RegisterExtension
     final TestTopologyExtension<Object, String> testTopology = new TestTopologyExtension<>(this.app.getTopology(),
-            this.app.getKafkaProperties());
+            WordCount.getKafkaProperties());
 
     @Test
     void shouldAggregateSameWordStream() {

fluent-kafka-streams-tests-junit5/src/test/java/com/bakdata/fluent_kafka_streams_tests/junit5/WordCountWitherTest.java

@@ -1,7 +1,7 @@
 /*
  * MIT License
  *
- * Copyright (c) 2019 bakdata GmbH
+ * 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
@@ -37,7 +37,7 @@ class WordCountWitherTest {
 
     @RegisterExtension
     final TestTopologyExtension<Object, String> testTopology =
-            new TestTopologyExtension<>(this.app::getTopology, this.app.getKafkaProperties())
+            new TestTopologyExtension<>(this.app::getTopology, WordCount.getKafkaProperties())
                     .withDefaultValueSerde(Serdes.String())
                     .withSchemaRegistryMock(new SchemaRegistryMock(List.of(new AvroSchemaProvider())));
 

fluent-kafka-streams-tests-junit5/src/test/java/com/bakdata/fluent_kafka_streams_tests/junit5/test_applications/WordCount.java

@@ -1,40 +1,57 @@
+/*
+ * 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.fluent_kafka_streams_tests.junit5.test_applications;
 
 import java.util.Arrays;
-import java.util.Properties;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.regex.Pattern;
 import lombok.Getter;
 import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.common.serialization.Serdes;
-import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.common.serialization.Serdes.StringSerde;
 import org.apache.kafka.streams.StreamsBuilder;
 import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.Topology;
 import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.kstream.KTable;
 import org.apache.kafka.streams.kstream.Produced;
 
+@Getter
 public class WordCount {
-    @Getter
     private final String inputTopic = "wordcount-input";
 
-    @Getter
     private final String outputTopic = "wordcount-output";
 
-    public static void main(final String[] args) {
-        final WordCount wordCount = new WordCount();
-        final KafkaStreams streams = new KafkaStreams(wordCount.getTopology(), wordCount.getKafkaProperties());
-        streams.start();
-        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
-    }
-
-    public Properties getKafkaProperties() {
+    public static Map<String, Object> getKafkaProperties() {
         final String brokers = "localhost:9092";
-        final Properties kafkaConfig = new Properties();
-        kafkaConfig.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount");
-        kafkaConfig.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
-        kafkaConfig.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
-        kafkaConfig.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
+        final Map<String, Object> kafkaConfig = new HashMap<>();
+        kafkaConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount");
+        kafkaConfig.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
+        kafkaConfig.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, StringSerde.class);
+        kafkaConfig.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, StringSerde.class);
         return kafkaConfig;
     }