Skip to content

Commit

Permalink
Use Map instead of Properties (#91)
Browse files Browse the repository at this point in the history
  • Loading branch information
@philipp94831
philipp94831 authored Jul 19, 2024
1 parent f6cb911 commit 84565da
Show file tree
Hide file tree
Showing 32 changed files with 550 additions and 285 deletions.
28 changes: 14 additions & 14 deletions ...-junit4/src/main/java/com/bakdata/fluent_kafka_streams_tests/junit4/TestTopologyRule.java
View file
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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;
Expand Down Expand Up @@ -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);
Expand All @@ -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,
Expand Down
2 changes: 1 addition & 1 deletion ...sts-junit4/src/test/java/com/bakdata/fluent_kafka_streams_tests/junit4/WordCountTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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() {
Expand Down
4 changes: 2 additions & 2 deletions ...t/java/com/bakdata/fluent_kafka_streams_tests/junit4/WordCountWithStaticTopologyTest.java
View file
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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() {
Expand Down
4 changes: 2 additions & 2 deletions ...nit4/src/test/java/com/bakdata/fluent_kafka_streams_tests/junit4/WordCountWitherTest.java
View file
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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())));

Expand Down
51 changes: 34 additions & 17 deletions .../test/java/com/bakdata/fluent_kafka_streams_tests/junit4/test_applications/WordCount.java
View file
Original file line number Diff line number Diff line change
@@ -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;
}

Expand Down
28 changes: 14 additions & 14 deletions ...t5/src/main/java/com/bakdata/fluent_kafka_streams_tests/junit5/TestTopologyExtension.java
View file
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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;
Expand Down Expand Up @@ -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);
Expand All @@ -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,
Expand Down
2 changes: 1 addition & 1 deletion ...sts-junit5/src/test/java/com/bakdata/fluent_kafka_streams_tests/junit5/WordCountTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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() {
Expand Down
4 changes: 2 additions & 2 deletions ...t/java/com/bakdata/fluent_kafka_streams_tests/junit5/WordCountWithStaticTopologyTest.java
View file
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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() {
Expand Down
4 changes: 2 additions & 2 deletions ...nit5/src/test/java/com/bakdata/fluent_kafka_streams_tests/junit5/WordCountWitherTest.java
View file
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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())));

Expand Down
51 changes: 34 additions & 17 deletions .../test/java/com/bakdata/fluent_kafka_streams_tests/junit5/test_applications/WordCount.java
View file
Original file line number Diff line number Diff line change
@@ -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;
}

Expand Down
Loading

0 comments on commit 84565da

Please sign in to comment.