Added asyn features

src/main/java/com/kafka/producer/demo/kafkaproducer/controller/KafkaRestController.java

@@ -9,6 +9,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
+import org.springframework.util.StringUtils;
 import org.springframework.web.bind.annotation.*;
 
 @Slf4j
@@ -28,6 +29,7 @@ public class KafkaRestController {
     @PostMapping(value = "/sendEmployeeData", consumes = MediaType.APPLICATION_JSON_VALUE)
     public ResponseEntity<String> sendEmployeeData(@RequestBody Employee employee) {
         //log.info("Employee :: " +employee);
+        if(!StringUtils.isEmpty(employee))
         publisher.sendEmployeeData(employee);
         return new ResponseEntity("Employee Data sent successfully.", HttpStatus.OK);
     }

src/main/java/com/kafka/producer/demo/kafkaproducer/service/KafkaMessagePublisher.java

 package com.kafka.producer.demo.kafkaproducer.service;
 
 import com.kafka.producer.demo.kafkaproducer.model.Employee;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.annotation.Required;
 import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
 import org.springframework.stereotype.Service;
+import org.springframework.util.concurrent.ListenableFuture;
+import org.springframework.util.concurrent.ListenableFutureCallback;
+
+import java.util.UUID;
 
 @Service
 public class KafkaMessagePublisher {
 
+    private static Logger LOGGER = LoggerFactory.getLogger(KafkaMessagePublisher.class);
+
     @Autowired
     private KafkaTemplate<String, String> kafkaTemplate;
 
@@ -17,14 +26,34 @@ public class KafkaMessagePublisher {
     @Qualifier("CustomKafkaTemplate")
     private KafkaTemplate<String, Employee> customKafkaTemplate;
 
-    String kafkaTopic = "testTopic";
+    String topicName = "testTopic";
 
     public void sendMessage(String message) {
-        kafkaTemplate.send(kafkaTopic, message);
+        //Fire and Forgot
+        //kafkaTemplate.send(topicName, message);
+        //Fire & Expect Result
+        ListenableFuture<SendResult<String, String>> future =
+                kafkaTemplate.send(topicName, UUID.randomUUID().toString(), message);
+
+        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
+
+            @Override
+            public void onSuccess(SendResult<String, String> result) {
+                LOGGER.info("Sent message=[" + message +
+                        "] with offset=[" + result.getRecordMetadata().offset() + "]");
+            }
+
+            @Override
+            public void onFailure(Throwable ex) {
+                LOGGER.info("Unable to send message=["
+                        + message + "] due to : " + ex.getMessage());
+            }
+        });
     }
 
     public void sendEmployeeData(Employee emp) {
-        customKafkaTemplate.send(kafkaTopic, emp);
+
+        customKafkaTemplate.send(topicName, emp);
     }
 
 }