close Kafka producer after sending the message

cognitree · Jul 4, 2019 · 19e8a4a · 19e8a4a
1 parent 321b9ec
commit 19e8a4a
Showing 1 changed file with 15 additions and 13 deletions.
diff --git a/...age-handler/src/main/java/com/cognitree/kronos/executor/handlers/KafkaMessageHandler.java b/...age-handler/src/main/java/com/cognitree/kronos/executor/handlers/KafkaMessageHandler.java
@@ -20,6 +20,7 @@
 import com.cognitree.kronos.executor.model.TaskResult;
 import com.cognitree.kronos.model.Task;
 import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import org.apache.kafka.clients.producer.KafkaProducer;
@@ -44,19 +45,17 @@ public class KafkaMessageHandler implements TaskHandler {
     // message to push to kafka topic
     private static final String MESSAGE_KEY = "message";
 
+    private JsonNode producerConfig;
     private String defaultTopic;
-    private KafkaProducer<String, String> kafkaProducer;
     private Task task;
 
     @Override
     public void init(Task task, ObjectNode config) {
-        logger.info("Initializing producer for kafka with config {}", config);
         if (config == null || !config.hasNonNull(KAFKA_PRODUCER_CONFIG_KEY)) {
             throw new IllegalArgumentException("missing mandatory configuration: [kafkaProducerConfig]");
         }
         this.task = task;
-        Properties kafkaProducerConfig = OBJECT_MAPPER.convertValue(config.get(KAFKA_PRODUCER_CONFIG_KEY), Properties.class);
-        kafkaProducer = new KafkaProducer<>(kafkaProducerConfig);
+        producerConfig = config.get(KAFKA_PRODUCER_CONFIG_KEY);
         defaultTopic = config.get(TOPIC_KEY).asText();
     }
 
@@ -76,14 +75,17 @@ public TaskResult execute() {
     }
 
     private void send(String topic, String record) {
-        logger.trace("Received request to send message {} to topic {}.", record, topic);
-        ProducerRecord<String, String> producerRecord =
-                new ProducerRecord<>(topic, record);
-        kafkaProducer.send(producerRecord, (metadata, exception) -> {
-            if (exception != null) {
-                logger.error("Error sending record {} over kafka to topic {}.",
-                        record, topic, exception);
-            }
-        });
+        logger.debug("Received request to send message {} to topic {}.", record, topic);
+        logger.debug("Initializing producer for kafka with config {}", producerConfig);
+        final Properties kafkaProducerConfig = OBJECT_MAPPER.convertValue(this.producerConfig, Properties.class);
+        try (final KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(kafkaProducerConfig)) {
+            final ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, record);
+            kafkaProducer.send(producerRecord, (metadata, exception) -> {
+                if (exception != null) {
+                    logger.error("Error sending record {} over kafka to topic {}.",
+                            record, topic, exception);
+                }
+            });
+        }
     }
 }