feat: Enable more Kafka logging and reduce timeout (#20)

feat: Enable Kafka network logging at warning level
to catch warnings relating to the brokers being
unavailable.

feat: Reduce the timeout for getting partitions data
to see errors more quickly and do no retry as
the app will try again due to to TimeoutStream.

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

Author: katheris

README.md

@@ -38,4 +38,6 @@ Test out the app by connecting to the websocket endpoints (UI will be added soon
 To configure the application to connect to your Kafka edit the properties file called `kafka.properties`.
 Alternatively you can provide a custom path to the properties file using `-Dproperties_path=<path>` when starting the application.
 
-If your Kafka is secured you will need to enable the security configuration options in your properties file
+If your Kafka is secured you will need to enable the security configuration options in your properties file.
+
+To increase the logging level to debug provide a system property at start time: `-Dlog.level=debug`.

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

@@ -12,6 +12,7 @@
 import io.vertx.core.json.JsonObject;
 import io.vertx.kafka.client.producer.KafkaProducer;
 import io.vertx.kafka.client.producer.KafkaProducerRecord;
+import org.apache.kafka.clients.producer.ProducerConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -35,9 +36,12 @@ public void start(Promise<Void> startPromise) {
   }
 
   private void setup(HashMap<String, String> props) {
+    // Don't retry and only wait 10 secs for partition info as this is a demo app
+    props.put(ProducerConfig.RETRIES_CONFIG, "0");
+    props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "10000");
     KafkaProducer<String, String> kafkaProducer = KafkaProducer.create(vertx, props);
 
-    kafkaProducer.exceptionHandler(err -> logger.error("Kafka error: {}", err));
+    kafkaProducer.exceptionHandler(err -> logger.debug("Kafka error: {}", err));
 
     TimeoutStream timerStream = vertx.periodicStream(2000);
     timerStream.handler(tick -> produceKafkaRecord(kafkaProducer, props.get(Main.TOPIC_KEY)));

src/main/resources/logback.xml

@@ -9,8 +9,9 @@
   <logger name="io.netty" level="warn"/>
   <logger name="io.vertx" level="info"/>
   <logger name="org.apache.kafka" level="error"/>
+  <logger name="org.apache.kafka.clients.NetworkClient" level="warn"/>
 
-  <root level="info">
+  <root level="${log.level:-info}">
     <appender-ref ref="STDOUT"/>
   </root>