Code & Setup für acks-Übung überarbeitet

[demos/kafka/training] / src / main / java / de / juplo / kafka / Application.java

diff --git a/src/main/java/de/juplo/kafka/Application.java b/src/main/java/de/juplo/kafka/Application.java
index 4632e22..2dc5756 100644 (file)
--- a/src/main/java/de/juplo/kafka/Application.java
+++ b/src/main/java/de/juplo/kafka/Application.java
@@ -10,6 +10,8 @@ import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.SendResult;
 import org.springframework.util.concurrent.ListenableFuture;
 
+import java.util.List;
+
 
 @Slf4j
 @SpringBootApplication
@@ -21,10 +23,17 @@ public class Application implements ApplicationRunner
   @Override
   public void run(ApplicationArguments args)
   {
+    List<String> keys = args.getOptionValues("key");
+    if (keys == null || keys.size() != 1)
+    {
+      log.error("Exactly one key has to be given with --key");
+      return;
+    }
+    String key = keys.get(0);
     for (int i = 0; true; i++)
     {
       ListenableFuture<SendResult<String, String>> listenableFuture =
-          kafkaTemplate.send("test", Long.toString(i%10), Long.toString(i));
+          kafkaTemplate.send("test", key, Long.toString(i));
 
       listenableFuture.addCallback(
           result -> log.info(