From 0de33af47b76eaad1a2218dc88ae8a53b0235979 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Mon, 13 May 2024 21:04:56 +0200 Subject: [PATCH] counter: 1.2.7 - Refined bean-naming in `CounterApplicationConfiguration` --- pom.xml | 2 +- .../CounterApplicationConfiguriation.java | 24 +++++++++---------- .../CounterStreamProcessorTopologyTest.java | 8 +++---- 3 files changed, 17 insertions(+), 17 deletions(-) diff --git a/pom.xml b/pom.xml index 46a7150..22b9da0 100644 --- a/pom.xml +++ b/pom.xml @@ -10,7 +10,7 @@ de.juplo.kafka.wordcount counter - 1.2.6 + 1.2.7 Wordcount-Counter Word-counting stream-processor of the multi-user wordcount-example diff --git a/src/main/java/de/juplo/kafka/wordcount/counter/CounterApplicationConfiguriation.java b/src/main/java/de/juplo/kafka/wordcount/counter/CounterApplicationConfiguriation.java index 409c035..9785f69 100644 --- a/src/main/java/de/juplo/kafka/wordcount/counter/CounterApplicationConfiguriation.java +++ b/src/main/java/de/juplo/kafka/wordcount/counter/CounterApplicationConfiguriation.java @@ -26,12 +26,12 @@ import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.St public class CounterApplicationConfiguriation { @Bean - public Properties propertyMap(CounterApplicationProperties properties) + public Properties streamProcessorProperties(CounterApplicationProperties counterProperties) { Properties propertyMap = new Properties(); - propertyMap.put(StreamsConfig.APPLICATION_ID_CONFIG, properties.getApplicationId()); - propertyMap.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServer()); + propertyMap.put(StreamsConfig.APPLICATION_ID_CONFIG, counterProperties.getApplicationId()); + propertyMap.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, counterProperties.getBootstrapServer()); propertyMap.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, JsonSerde.class.getName()); propertyMap.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class.getName()); propertyMap.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false); @@ -40,10 +40,10 @@ public class CounterApplicationConfiguriation propertyMap.put(JsonDeserializer.VALUE_DEFAULT_TYPE, Word.class.getName()); propertyMap.put(JsonDeserializer.USE_TYPE_INFO_HEADERS, false); propertyMap.put(StreamsConfig.STATE_DIR_CONFIG, "target"); - if (properties.getCommitInterval() != null) - propertyMap.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, properties.getCommitInterval()); - if (properties.getCacheMaxBytes() != null) - propertyMap.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, properties.getCacheMaxBytes()); + if (counterProperties.getCommitInterval() != null) + propertyMap.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, counterProperties.getCommitInterval()); + if (counterProperties.getCacheMaxBytes() != null) + propertyMap.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, counterProperties.getCacheMaxBytes()); propertyMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); return propertyMap; @@ -51,15 +51,15 @@ public class CounterApplicationConfiguriation @Bean(initMethod = "start", destroyMethod = "stop") public CounterStreamProcessor streamProcessor( - CounterApplicationProperties properties, - Properties propertyMap, + CounterApplicationProperties applicationProperties, + Properties streamProcessorProperties, KeyValueBytesStoreSupplier storeSupplier, ConfigurableApplicationContext context) { CounterStreamProcessor streamProcessor = new CounterStreamProcessor( - properties.getInputTopic(), - properties.getOutputTopic(), - propertyMap, + applicationProperties.getInputTopic(), + applicationProperties.getOutputTopic(), + streamProcessorProperties, storeSupplier); streamProcessor.streams.setUncaughtExceptionHandler((Throwable e) -> diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java b/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java index 2c8bd1e..fe85dc4 100644 --- a/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java +++ b/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java @@ -26,12 +26,12 @@ public class CounterStreamProcessorTopologyTest OUT, Stores.inMemoryKeyValueStore("TOPOLOGY-TEST")); - CounterApplicationConfiguriation config = + CounterApplicationConfiguriation applicationConfiguriation = new CounterApplicationConfiguriation(); - Properties properties = - config.propertyMap(new CounterApplicationProperties()); + Properties streamProcessorProperties = + applicationConfiguriation.streamProcessorProperties(new CounterApplicationProperties()); - TopologyTestDriver testDriver = new TopologyTestDriver(topology, properties); + TopologyTestDriver testDriver = new TopologyTestDriver(topology, streamProcessorProperties); TestInputTopic in = testDriver.createInputTopic( IN, -- 2.20.1