splitter: 1.0.0-vanilla-kafka - Fixed the test "Context Loads"

[demos/kafka/wordcount] / src / main / java / de / juplo / kafka / wordcount / splitter / SplitterStreamProcessor.java

diff --git a/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterStreamProcessor.java b/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterStreamProcessor.java
index 791e164..191428c 100644 (file)
--- a/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterStreamProcessor.java
+++ b/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterStreamProcessor.java
@@ -9,8 +9,7 @@ import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.WakeupException;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
-import org.springframework.boot.ApplicationArguments;
-import org.springframework.boot.ApplicationRunner;
+import org.springframework.core.task.TaskExecutor;
 import org.springframework.stereotype.Component;
 import org.springframework.util.Assert;
 
@@ -26,7 +25,7 @@ import java.util.stream.Collectors;
 
 @Component
 @Slf4j
-public class SplitterStreamProcessor implements ApplicationRunner
+public class SplitterStreamProcessor implements Runnable
 {
   final static Pattern PATTERN = Pattern.compile("\\W+");
 
@@ -45,7 +44,8 @@ public class SplitterStreamProcessor implements ApplicationRunner
 
   public SplitterStreamProcessor(
       SplitterApplicationProperties properties,
-      Clock clock)
+      Clock clock,
+      TaskExecutor executor)
   {
     this.inputTopic = properties.getInputTopic();
     this.outputTopic = properties.getOutputTopic();
@@ -72,10 +72,11 @@ public class SplitterStreamProcessor implements ApplicationRunner
     props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
     props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
     producer = new KafkaProducer<>(props);
+
+    executor.execute(this);
   }
 
-  @Override
-  public void run(ApplicationArguments args)
+  public void run()
   {
     running.lock();