feat: Make topic configurable (#12)

Signed-off-by: Katherine Stanley <katheris@uk.ibm.com>

Author: katheris

README.md

@@ -2,7 +2,7 @@
 
 A starter app for testing out your [Apache Kafka](https://kafka.apache.org) deployment and trying out the [Vert.x Kafka client](https://vertx.io/docs/vertx-kafka-client/java/).
 
-The app allows you to send records to a topic in Kafka called `test` every two seconds and consume back those records.
+The app allows you to send records to a topic in Kafka called `demo` every two seconds and consume back those records.
 
 ## Running the app
 

src/main/java/kafka/vertx/demo/PeriodicProducer.java

@@ -20,6 +20,7 @@
 
 import java.time.Instant;
 import java.util.HashMap;
+import java.util.Optional;
 
 public class PeriodicProducer extends AbstractVerticle {
 
@@ -44,10 +45,12 @@ private void setup(JsonObject config, Promise<Void> startPromise) {
     config.forEach(entry -> props.put(entry.getKey(), entry.getValue().toString()));
     KafkaProducer<String, String> kafkaProducer = KafkaProducer.create(vertx, props);
 
+    String topic = Optional.ofNullable(config.getString("topic")).orElse(Main.TOPIC);
+
     kafkaProducer.exceptionHandler(err -> logger.error("Kafka error: {}", err));
 
     TimeoutStream timerStream = vertx.periodicStream(2000);
-    timerStream.handler(tick -> produceKafkaRecord(kafkaProducer));
+    timerStream.handler(tick -> produceKafkaRecord(kafkaProducer, topic));
     timerStream.pause();
 
     vertx.eventBus().<JsonObject>consumer(Main.PERIODIC_PRODUCER_ADDRESS, message -> handleCommand(timerStream, message));
@@ -67,10 +70,9 @@ private void handleCommand(TimeoutStream timerStream, Message<JsonObject> messag
     }
   }
 
-  private void produceKafkaRecord(KafkaProducer<String, String> kafkaProducer) {
+  private void produceKafkaRecord(KafkaProducer<String, String> kafkaProducer, String topic) {
     String payload = customMessage;
-
-    KafkaProducerRecord<String, String> record = KafkaProducerRecord.create(Main.TOPIC, payload);
+    KafkaProducerRecord<String, String> record = KafkaProducerRecord.create(topic, payload);
 
     kafkaProducer
       .send(record)

src/main/java/kafka/vertx/demo/WebSocketServer.java

@@ -22,6 +22,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.HashMap;
+import java.util.Optional;
 
 public class WebSocketServer extends AbstractVerticle {
 
@@ -120,11 +121,13 @@ private void handleConsumeSocket(ServerWebSocket webSocket) {
       vertx.eventBus().send(webSocket.textHandlerID(), payload.encode());
     });
 
-    kafkaConsumer.subscribe(Main.TOPIC)
-      .onSuccess(ok -> logger.info("Subscribed to {}", Main.TOPIC))
-      .onFailure(err -> logger.error("Could not subscribe to {}", Main.TOPIC, err));
+    String topic = Optional.ofNullable(kafkaConfig.get("topic")).orElse(Main.TOPIC);
 
-    TopicPartition partition = new TopicPartition().setTopic(Main.TOPIC);
+    kafkaConsumer.subscribe(topic)
+      .onSuccess(ok -> logger.info("Subscribed to {}", topic))
+      .onFailure(err -> logger.error("Could not subscribe to {}", topic, err));
+
+    TopicPartition partition = new TopicPartition().setTopic(topic);
 
     webSocket.handler(buffer -> {
       String action = buffer.toJsonObject().getString("action", "none");

src/main/resources/kafka.properties

@@ -1,4 +1,5 @@
 bootstrap.servers=localhost:9092
+# topic=
 
 ## Consumer specific properties
 group.id=my_group