used the library for Apache kafka and change the code accordinglly

src/main/java/com/safeway/epe/offersdataproducer/appconfiguration/ProducerConfiguration.java

@@ -4,32 +4,24 @@ import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.kafka.core.DefaultKafkaProducerFactory;
-import org.springframework.kafka.core.KafkaTemplate;
-import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.stereotype.Component;
 
-import java.util.HashMap;
-import java.util.Map;
+import java.util.Properties;
 
-@Configuration
+@Component
 public class ProducerConfiguration {
     public static final Logger logger = LoggerFactory.getLogger(ProducerConfiguration.class);
-    @Bean
-    public ProducerFactory<String, String> producerFactory() {
-        logger.info("Setting producer configuration ");
-        Map<String, Object> config = new HashMap<>();
-        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, AppConfigs.BOOTSTRAP_SERVER);
-        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
-        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
-        return new DefaultKafkaProducerFactory<>(config);
-    }
-    @Bean
-    @Qualifier("kafkaTemplate")
-    public KafkaTemplate<String, String> kafkaTemplate()
-    {
-        return new KafkaTemplate<>(producerFactory());
+
+
+    public Properties configuration(){
+        Properties properties = new Properties();
+        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG , AppConfigs.BOOTSTRAP_SERVER);
+        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG , StringSerializer.class.getName());
+        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG , StringSerializer.class.getName());
+
+       return properties;
     }
-}
+
+
+
+   }

src/main/java/com/safeway/epe/offersdataproducer/producer/OffersDataProducer.java

@@ -3,27 +3,33 @@ package com.safeway.epe.offersdataproducer.producer;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.safeway.epe.offersdataproducer.appconfiguration.AppConfigs;
+import com.safeway.epe.offersdataproducer.appconfiguration.ProducerConfiguration;
 import com.safeway.epe.offersdataproducer.dtomodel.OffersDataDTO;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.kafka.core.KafkaTemplate;
-import org.springframework.kafka.support.SendResult;
 import org.springframework.stereotype.Component;
-import org.springframework.util.concurrent.ListenableFuture;
+
+import java.util.concurrent.Future;
 
 @Component
 public class OffersDataProducer {
     private static final Logger logger = LoggerFactory.getLogger(OffersDataProducer.class);
     @Autowired
-    private KafkaTemplate<String, String> kafkaTemplate;
+    private ProducerConfiguration producerConfiguration;
     @Autowired
     private ObjectMapper objectMapper;
-    public ListenableFuture<SendResult<String, String>> sendMessage(OffersDataDTO offersDataDTO) throws JsonProcessingException {
-        logger.info("Start :: OffersDataProducer.sendMessage");
+
+    public Future sendEvent(OffersDataDTO offersDataDTO) throws JsonProcessingException {
+        logger.info("Start :: OffersDataProducer.sendEvent");
         String offersDataDTOResp = objectMapper.writeValueAsString(offersDataDTO);
-        ListenableFuture<SendResult<String, String>> message=  kafkaTemplate.send(AppConfigs.TOPIC_NAME, offersDataDTOResp);
-        logger.info("End :: OffersDataProducer.sendMessage");
-        return message;
+        KafkaProducer producer =  new KafkaProducer(producerConfiguration.configuration());
+        ProducerRecord record = new ProducerRecord(AppConfigs.TOPIC_NAME , offersDataDTOResp);
+        Future<RecordMetadata> future = producer.send(record);
+        logger.info("End :: OffersDataProducer.sendEvent");
+        return future;
     }
 }

src/main/java/com/safeway/epe/offersdataproducer/resource/OffersDataProducerResource.java

@@ -4,13 +4,14 @@ import com.fasterxml.jackson.core.JsonProcessingException;
 import com.safeway.epe.offersdataproducer.appconfiguration.AppConfigs;
 import com.safeway.epe.offersdataproducer.domodel.OffersDataDO;
 import com.safeway.epe.offersdataproducer.service.OffersDataProducerService;
+import org.apache.kafka.clients.producer.RecordMetadata;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
-import org.springframework.kafka.support.SendResult;
-import org.springframework.util.concurrent.ListenableFuture;
 import org.springframework.web.bind.annotation.*;
 
+import java.util.concurrent.Future;
+
 @RestController
 @RequestMapping(AppConfigs.CLASS_LEVEL_PRODUCER_URL)
 public class OffersDataProducerResource {
@@ -18,8 +19,8 @@ public class OffersDataProducerResource {
     private OffersDataProducerService offersDataProducerService;
     @PostMapping(AppConfigs.CLASS_LEVEL_PRODUCER_URL)
     @ResponseStatus(HttpStatus.CREATED)
-    public ResponseEntity<ListenableFuture<SendResult<String, String>>> publishOfferEvent(@RequestBody OffersDataDO offersDataDO) throws JsonProcessingException {
-        ListenableFuture<SendResult<String, String>> offerEventStatus = offersDataProducerService.publishOfferEvent(offersDataDO);
+    public ResponseEntity<Future<RecordMetadata>> publishOfferEvent(@RequestBody OffersDataDO offersDataDO) throws JsonProcessingException {
+        Future<RecordMetadata> offerEventStatus = offersDataProducerService.publishOfferEvent(offersDataDO);
         return ResponseEntity.status(HttpStatus.CREATED).body(offerEventStatus);
     }
 }

src/main/java/com/safeway/epe/offersdataproducer/service/OffersDataProducerService.java

@@ -2,9 +2,9 @@ package com.safeway.epe.offersdataproducer.service;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.safeway.epe.offersdataproducer.domodel.OffersDataDO;
-import org.springframework.kafka.support.SendResult;
-import org.springframework.util.concurrent.ListenableFuture;
+
+import java.util.concurrent.Future;
 
 public interface OffersDataProducerService {
-    ListenableFuture<SendResult<String, String>> publishOfferEvent(OffersDataDO offersDataDO) throws JsonProcessingException;
+    public Future publishOfferEvent(OffersDataDO offersDataDO) throws JsonProcessingException;
 }

src/main/java/com/safeway/epe/offersdataproducer/service/impl/OffersDataProducerServiceImpl.java

@@ -8,13 +8,17 @@ import com.safeway.epe.offersdataproducer.producer.OffersDataProducer;
 import com.safeway.epe.offersdataproducer.service.OffersDataProducerService;
 import com.safeway.epe.offersdataproducer.transformer.OffersDataBoToDTO;
 import com.safeway.epe.offersdataproducer.transformer.OffersDataDoToBo;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.kafka.support.SendResult;
 import org.springframework.stereotype.Service;
-import org.springframework.util.concurrent.ListenableFuture;
+
+import java.util.concurrent.Future;
 
 @Service
 public class OffersDataProducerServiceImpl implements OffersDataProducerService {
+    public static final Logger logger = LoggerFactory.getLogger(OffersDataProducerServiceImpl.class);
     @Autowired
     private OffersDataBoToDTO offersDataBoToDTO;
     @Autowired
@@ -22,10 +26,12 @@ public class OffersDataProducerServiceImpl implements OffersDataProducerService
     @Autowired
     private OffersDataProducer offersDataProducer;
     @Override
-    public ListenableFuture<SendResult<String, String>> publishOfferEvent(OffersDataDO offersDataDO) throws JsonProcessingException {
+    public Future publishOfferEvent(OffersDataDO offersDataDO) throws JsonProcessingException {
+        logger.info("Start :: OffersDataProducerServiceImpl.publishOfferEvent");
         OffersDataBO offersDataBO = offersDataDoToBo.transerToBo(offersDataDO);
         OffersDataDTO offersDataDTO = offersDataBoToDTO.transferToDTO(offersDataBO);
-        ListenableFuture<SendResult<String, String>> eventStatus = offersDataProducer.sendMessage(offersDataDTO);
-        return eventStatus;
+        Future<RecordMetadata> future = offersDataProducer.sendEvent(offersDataDTO);
+        logger.info("End :: OffersDataProducerServiceImpl.publishOfferEvent");
+        return future;
     }
 }

src/test/java/com/safeway/epe/offersdataproducer/resource/OffersDataDTO.json

-{
-  "offerId": 135918555,
+{ "offerId": 135918555, "offersDTO": [ { "infoDTO": { "offerId": 135918555, "idDTO": { "offerId": 135918555, "manufacturerId": "MMM" }, "offerProgramCode": "SC", "terminalsDTO": [ "Groceryworks (Safeway.com)", "Smart Cart (QVS Virtual Terminals)", "Bakery" ] }, "rulesDTO": { "offerId": 135918555, "startDate": "2019-07-27T07:00:00.000+00:00", "endDate": "2021-02-27T06:59:58.000+00:00", "benefitDTO": { "benefitValueType": "A", "discountDTO": [ { "allowNegative": false, "flexNegative": false, "discountTierDTO": [ { "amount": 6.99, "upTo": "3", "itemLimit": 5, "weightLimit": 2 } ] } ], "pointsDTO": [ { "generalPoints": 5, "loyaltyPoints": 250 } ] } } } ] }
\ No newline at end of file
-  "offersDTO": [
-    {
-      "infoDTO":
-      {
-        "offerId": 135918555,
-        "idDTO":
-        {
-          "offerId": 135918555,
-          "manufacturerId": "MMM"
-        },
-        "offerProgramCode": "SC",
-        "terminalsDTO":
-        [
-          "Groceryworks (Safeway.com)",
-          "Smart Cart (QVS Virtual Terminals)",
-          "Bakery"
-        ]
-      },
-      "rulesDTO":
-      {
-        "offerId": 135918555,
-        "startDate": "2019-07-27T07:00:00.000+00:00",
-        "endDate": "2021-02-27T06:59:58.000+00:00",
-        "benefitDTO":
-        {
-          "benefitValueType": "A",
-          "discountDTO":
-          [
-            {
-              "allowNegative": false,
-              "flexNegative": false,
-              "discountTierDTO":
-              [
-                {
-                  "amount": 6.99,
-                  "upTo": "3",
-                  "itemLimit": 5,
-                  "weightLimit": 2
-                }
-              ]
-            }
-          ],
-          "pointsDTO":
-          [
-            {
-              "generalPoints": 5,
-              "loyaltyPoints": 250
-            }
-          ]
-        }
-      }
-    }
-  ]
-}
\ No newline at end of file

src/test/java/com/safeway/epe/offersdataproducer/resource/OffersDataProducerResourceComponentTest.java

 package com.safeway.epe.offersdataproducer.resource;
-
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.Lists;
 import com.safeway.epe.offersdataproducer.appconfiguration.AppConfigs;
 import com.safeway.epe.offersdataproducer.domodel.*;
 import com.safeway.epe.offersdataproducer.dtomodel.*;
+import com.safeway.epe.offersdataproducer.producer.OffersDataProducer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.TopicPartition;
@@ -13,7 +13,6 @@ import org.json.JSONException;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.ArgumentCaptor;
-import org.mockito.ArgumentMatchers;
 import org.mockito.Mockito;
 import org.skyscreamer.jsonassert.Customization;
 import org.skyscreamer.jsonassert.JSONAssert;
@@ -22,20 +21,15 @@ import org.skyscreamer.jsonassert.comparator.CustomComparator;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.test.mock.mockito.MockBean;
-import org.springframework.boot.test.web.client.TestRestTemplate;
 import org.springframework.http.ResponseEntity;
-import org.springframework.kafka.core.KafkaTemplate;
-import org.springframework.kafka.support.SendResult;
 import org.springframework.test.context.ActiveProfiles;
 import org.springframework.test.context.junit4.SpringRunner;
-import org.springframework.util.concurrent.ListenableFuture;
 import org.springframework.util.concurrent.SettableListenableFuture;
-
 import java.io.BufferedReader;
 import java.io.FileReader;
 import java.io.IOException;
 import java.util.concurrent.ExecutionException;
-
+import java.util.concurrent.Future;
 @RunWith(SpringRunner.class)
 @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
 @ActiveProfiles("test")
@@ -44,72 +38,26 @@ public class OffersDataProducerResourceComponentTest {
     @Autowired
     private OffersDataProducerResource offersDataProducerResource;
     @MockBean
-    private KafkaTemplate kafkaTemplate;
+    private OffersDataProducer offersDataProducer;
     @Autowired
-    private TestRestTemplate testRestTemplate;
-    @Test
-    public void publishOfferEventTest() throws JsonProcessingException, ExecutionException, InterruptedException {
-        OffersDataDO offersDataDO = buildOffersDataDO();
-        OffersDataDTO offersDataDTO = buildOffersDataDTO();
-        ObjectMapper mapper = new ObjectMapper();
-        String excpected = mapper.writeValueAsString(offersDataDTO);
-        SettableListenableFuture future = new SettableListenableFuture();
-        ProducerRecord<Integer, String> producerRecord = new ProducerRecord(AppConfigs.TOPIC_NAME, null, offersDataDTO);
-        RecordMetadata recordMetadata = new RecordMetadata(new TopicPartition(AppConfigs.TOPIC_NAME, 0),
-                8, 1, 342, System.currentTimeMillis(), -1, 581);
-        SendResult<Integer, String> sendResult = new SendResult<Integer, String>(producerRecord, recordMetadata);
-        future.set(sendResult);
-        Mockito.when(kafkaTemplate.send(Mockito.anyString(),Mockito.any())).thenReturn(future);
-        ResponseEntity<ListenableFuture<SendResult<String, String>>> result = offersDataProducerResource.publishOfferEvent(offersDataDO);
-        String actual = mapper.writeValueAsString(result.getBody().get().getProducerRecord().value());
-        JSONAssert.assertEquals(excpected ,actual ,false);
-    }
+    private  ObjectMapper objectMapper;
     @Test
     public void publishOfferEventWithArgumentCatureTest() throws JsonProcessingException, ExecutionException, InterruptedException {
         ObjectMapper mapper = new ObjectMapper();
         OffersDataDO offersDataDO = buildOffersDataDO();
         OffersDataDTO offersDataDTO = buildOffersDataDTO();
-        SettableListenableFuture future = new SettableListenableFuture();
-        ProducerRecord<Integer, String> producerRecord = new ProducerRecord(AppConfigs.TOPIC_NAME, null, offersDataDTO);
+        String offersDataDTOResp = mapper.writeValueAsString(offersDataDTO);
         RecordMetadata recordMetadata = new RecordMetadata(new TopicPartition(AppConfigs.TOPIC_NAME, 0),
-                8, 1, 342, System.currentTimeMillis(), -1, 581);
-        SendResult<Integer, String> sendResult = new SendResult<Integer, String>(producerRecord, recordMetadata);
-        future.set(sendResult);
+                8, 1, 342, 1111111L, -1, 581);
+        SettableListenableFuture future = new SettableListenableFuture();
+        future.set(recordMetadata);
         ArgumentCaptor<OffersDataDTO> serviceRequestArgumentCaptor = ArgumentCaptor.forClass(OffersDataDTO.class);
-        Mockito.when(kafkaTemplate.send(ArgumentMatchers.any(),serviceRequestArgumentCaptor.capture())).thenReturn(future);
-        ResponseEntity<ListenableFuture<SendResult<String, String>>> result = offersDataProducerResource.publishOfferEvent(offersDataDO);
-        String actual = mapper.writeValueAsString(result.getBody().get().getProducerRecord().value());
-        validateRequest(serviceRequestArgumentCaptor ,"OffersDataDTO.json");
-        validteResponse(actual, "OffersDataDTO.json");
-    }
-    void validteResponse(String actualResponse, String fileName)
-    {
-        String expectedResponse = retrieveExpectedResult(fileName);
-        try {
-            JSONAssert.assertEquals(expectedResponse, actualResponse,
-                    new CustomComparator(JSONCompareMode.STRICT,
-                            new Customization(
-                                    "offerId" ,(value1 , value2) -> value1.equals(value2)) ,
-                            new Customization("offersDTO[0].infoDTO.idDTO.offerId" ,(value1 , value2) -> value1.equals(value2) ),
-                            new Customization("offersDTO[0].infoDTO.offerId" ,(value1 , value2) -> value1.equals(value2) ),
-                            new Customization("offersDTO[0].rulesDTO.benefitDTO.benefitValueType" ,(value1 , value2) -> value1.equals(value2) ),
-                            new Customization("offersDTO[0].rulesDTO.benefitDTO.discountDTO[0].allowNegative" ,(value1 , value2) -> value1.equals(value2) ),
-                            new Customization("offersDTO[0].rulesDTO.benefitDTO.discountDTO[0].discountTierDTO[0].upTo" ,(value1 , value2) -> value1.equals(value2) ),
-                            new Customization("offersDTO[0].rulesDTO.benefitDTO.discountDTO[0].flexNegative" ,(value1 , value2) -> value1.equals(value2) ),
-                            new Customization("offersDTO[0].rulesDTO.endDate" ,(value1 , value2) -> true ),
-                            new Customization("offersDTO[0].rulesDTO.offerId" ,(value1 , value2) -> value1.equals(value2) ),
-                            new Customization("offersDTO[0].rulesDTO.startDate" ,(value1 , value2) -> true ),
-                            new Customization("offersDTO[0].infoDTO.idDTO.offerId" ,(value1 , value2) -> value1.equals(value2) ),
-                            new Customization("offersDTO[0].infoDTO.offerId" ,(value1 , value2) -> value1.equals(value2) ),
-                            new Customization("offersDTO[0].rulesDTO.benefitDTO.benefitValueType" ,(value1 , value2) -> value1.equals(value2) )
-
-                    )
-            );
-        } catch (JSONException e) {
-            e.printStackTrace();
-        }
+        ProducerRecord record = new ProducerRecord(AppConfigs.TOPIC_NAME ,0, 1 ,offersDataDTOResp);
+        Mockito.when(offersDataProducer.sendEvent(serviceRequestArgumentCaptor.capture())).thenReturn(future);
+        ResponseEntity<Future<RecordMetadata>> result = offersDataProducerResource.publishOfferEvent(offersDataDO);
+        OffersDataDTO inputDataResp = serviceRequestArgumentCaptor.getValue();
+        validateRequest( serviceRequestArgumentCaptor,"OffersDataDTO.json");
     }
-
     private String retrieveExpectedResult(String fileName) {
         StringBuilder sb = new StringBuilder();
         try (BufferedReader br = new BufferedReader(new FileReader(FILE_PATH + fileName))) {
@@ -122,15 +70,13 @@ public class OffersDataProducerResourceComponentTest {
         }
         return sb.toString();
     }
-
-    private void validateRequest(ArgumentCaptor<OffersDataDTO> argumentCaptor, String fileName) {
-        String actual = String.valueOf(argumentCaptor.getValue());
+    private void validateRequest(ArgumentCaptor<OffersDataDTO> argumentCaptor, String fileName) throws JsonProcessingException {
+        String actual = objectMapper.writeValueAsString(argumentCaptor.getValue());
         String expectedServiceRequest = retrieveExpectedResult(fileName);
         try {
-            JSONAssert.assertEquals(expectedServiceRequest, actual,
+            JSONAssert.assertEquals(actual,expectedServiceRequest,
                     new CustomComparator(JSONCompareMode.STRICT,
-                            new Customization(
-                                    "offerId" ,(value1 , value2) -> value1.equals(value2)) ,
+                            new Customization("offerId" ,(value1 , value2) -> value1.equals(value2)) ,
                             new Customization("offersDTO[0].infoDTO.idDTO.offerId" ,(value1 , value2) -> value1.equals(value2) ),
                             new Customization("offersDTO[0].infoDTO.offerId" ,(value1 , value2) -> value1.equals(value2) ),
                             new Customization("offersDTO[0].rulesDTO.benefitDTO.benefitValueType" ,(value1 , value2) -> value1.equals(value2) ),
@@ -143,15 +89,12 @@ public class OffersDataProducerResourceComponentTest {
                             new Customization("offersDTO[0].infoDTO.idDTO.offerId" ,(value1 , value2) -> value1.equals(value2) ),
                             new Customization("offersDTO[0].infoDTO.offerId" ,(value1 , value2) -> value1.equals(value2) ),
                             new Customization("offersDTO[0].rulesDTO.benefitDTO.benefitValueType" ,(value1 , value2) -> value1.equals(value2) )
-
                     )
             );
         } catch (JSONException e) {
             e.printStackTrace();
         }
-
     }
-
     private OffersDataDO buildOffersDataDO() {
         DiscountTierDO discountTierDO = DiscountTierDO
                 .builder().amount(6.99f).upTo("3")

src/test/java/com/safeway/epe/offersdataproducer/resource/OffersDataProducerResourceTest.java

-package com.safeway.epe.offersdataproducer.resource;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.collect.Lists;
-import com.safeway.epe.offersdataproducer.appconfiguration.AppConfigs;
-import com.safeway.epe.offersdataproducer.domodel.*;
-import com.safeway.epe.offersdataproducer.service.OffersDataProducerService;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.clients.producer.RecordMetadata;
-import org.apache.kafka.common.TopicPartition;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.InjectMocks;
-import org.mockito.Mock;
-import org.mockito.Mockito;
-import org.skyscreamer.jsonassert.JSONAssert;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.boot.test.web.client.TestRestTemplate;
-import org.springframework.http.ResponseEntity;
-import org.springframework.kafka.support.SendResult;
-import org.springframework.test.context.ActiveProfiles;
-import org.springframework.test.context.junit4.SpringRunner;
-import org.springframework.util.concurrent.ListenableFuture;
-import org.springframework.util.concurrent.SettableListenableFuture;
-
-import java.util.concurrent.ExecutionException;
-
-@RunWith(SpringRunner.class)
-@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) // for restTemplate
-@ActiveProfiles("test")
-public class OffersDataProducerResourceTest {
-    @InjectMocks
-    private OffersDataProducerResource offersDataProducerResource;
-    @Mock
-    private OffersDataProducerService offersDataProducerService;
-    @Autowired
-    private TestRestTemplate testRestTemplate;
-    @Test
-    public void publishOfferEventTest() throws JsonProcessingException, ExecutionException, InterruptedException {
-        OffersDataDO offersDataDO = buildOffersDataDO();
-        ObjectMapper mapper = new ObjectMapper();
-        String excpected = mapper.writeValueAsString(offersDataDO);
-        SettableListenableFuture future = new SettableListenableFuture();
-        ProducerRecord<Integer, String> producerRecord = new ProducerRecord(AppConfigs.TOPIC_NAME, null, offersDataDO);
-        RecordMetadata recordMetadata = new RecordMetadata(new TopicPartition(AppConfigs.TOPIC_NAME, 0),
-                8, 1, 342, System.currentTimeMillis(), -1, 581);
-        SendResult<Integer, String> sendResult = new SendResult<Integer, String>(producerRecord, recordMetadata);
-
-        future.set(sendResult);
-        Mockito.when(offersDataProducerService.publishOfferEvent(Mockito.any(OffersDataDO.class)))
-                .thenReturn(future);
-        ResponseEntity<ListenableFuture<SendResult<String, String>>> result = offersDataProducerResource.publishOfferEvent(offersDataDO);
-        SendResult<String, String> sendResult1 = result.getBody().get();
-        assert sendResult1.getRecordMetadata().partition() == 0;
-        String actual = mapper.writeValueAsString(result.getBody().get().getProducerRecord().value());
-        JSONAssert.assertEquals(excpected ,actual ,false);
-    }
-    private OffersDataDO buildOffersDataDO() {
-        DiscountTierDO discountTierDO = DiscountTierDO
-                .builder().amount(6.99f).upTo("3")
-                .itemLimit(5).weightLimit(2).build();
-        PointsDO pointsDO = PointsDO.builder().generalPoints(5).loyaltyPoints(250).build();
-        DiscountDO discountDO = DiscountDO
-                .builder().discountTierDO(Lists.newArrayList(discountTierDO))
-                .allowNegative(false).flexNegative(false).build();
-        BenefitDO benefitDO = BenefitDO
-                .builder().discountDO(Lists.newArrayList(discountDO))
-                .pointsDO(Lists.newArrayList(pointsDO)).benefitValueType("A")
-                .build();
-        IdDO idDO = IdDO.builder().manufacturerId("MMM").offerId(135918444).build();
-        InfoDO infoDO = InfoDO
-                .builder().offerId(135918444).idDO(idDO).offerProgramCode("SC")
-                .terminalsDO(Lists.newArrayList("Groceryworks (Safeway.com)",
-                        "Smart Cart (QVS Virtual Terminals)",
-                        "Bakery")).build();
-        RulesDO rulesDO = RulesDO.builder().offerId(135918444).benefitDO(benefitDO).endDate("2021-02-27T06:59:58.000+00:00")
-                .startDate("2019-07-27T07:00:00.000+00:00").build();
-        OffersDO offersDO = OffersDO.builder().infoDO(infoDO).rulesDO(rulesDO).build();
-        OffersDataDO offersDataDO = OffersDataDO.builder().offerId(135918444).offersDO(Lists.newArrayList(offersDO)).build();
-
-        return offersDataDO;
-    }
-
-}