Repository: authorjapps/zerocode
Pull request #395 (2020-05-12 13:31:21)

Iss 330 send headers while producing to kafka
# send headers while producing to kafka

PR Branch
https://github.com/Cyril-Edme/zerocode/tree/ISS-330-send-headers-while-producing-to-kafka

## Motivation and Context
We should be able to send some headers in our Kafka message. As well as the consumer should be able to assert on the received headers.

## Checklist:

* [x] Unit tests added

* [x] Integration tests added

* [x] Test names are meaningful

* [x] Feature manually tested

* [x] Branch build passed

* [x] No 'package.*' in the imports

* [ ] Relevant Wiki page updated with clear instruction for the end user
  * [ ] Not applicable. This was only a refactor change, no functional or behaviour changes were introduced

* [ ] Http test added to `http-testing` module(if applicable) ?
  * [x] Not applicable. The changes did not affect HTTP automation flow

* [x] Kafka test added to `kafka-testing` module(if applicable) ?
  * [ ] Not applicable. The changes did not affect Kafka automation flow



Patch:
diff --git a/core/src/main/java/org/jsmart/zerocode/core/di/provider/GsonSerDeProvider.java b/core/src/main/java/org/jsmart/zerocode/core/di/provider/GsonSerDeProvider.java
index ff7546ef..e2e94b66 100644
--- a/core/src/main/java/org/jsmart/zerocode/core/di/provider/GsonSerDeProvider.java
+++ b/core/src/main/java/org/jsmart/zerocode/core/di/provider/GsonSerDeProvider.java
@@ -1,16 +1,79 @@
 package org.jsmart.zerocode.core.di.provider;
 
 import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.TypeAdapter;
+import com.google.gson.TypeAdapterFactory;
+import com.google.gson.reflect.TypeToken;
+import com.google.gson.stream.JsonReader;
+import com.google.gson.stream.JsonToken;
+import com.google.gson.stream.JsonWriter;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeaders;
 
 import javax.inject.Provider;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
 
 public class GsonSerDeProvider implements Provider<Gson> {
 
     @Override
     public Gson get() {
+        return new GsonBuilder()
+                .registerTypeAdapterFactory(KafkaHeadersAdapter.FACTORY)
+                .create();
+    }
+
+    static class KafkaHeadersAdapter extends TypeAdapter<Headers> {
+
+        static final TypeAdapterFactory FACTORY = new TypeAdapterFactory() {
+            @SuppressWarnings("unchecked")
+            @Override
+            public <T> TypeAdapter<T> create(Gson gson, TypeToken<T> type) {
+                if (type.getRawType() == Headers.class) {
+                    return (TypeAdapter<T>) new KafkaHeadersAdapter(gson);
+                }
+                return null;
+            }
+        };
+
+        private final Gson gson;
+
+        public KafkaHeadersAdapter(Gson gson) {
+            this.gson = gson;
+        }
+
+        @Override
+        public void write(JsonWriter writer, Headers value) throws IOException {
+            if (value == null || !value.iterator().hasNext()) {
+                writer.nullValue();
+            } else {
+                Map<String, String> headers = new HashMap<>();
+                value.forEach(header -> headers.put(header.key(), new String(header.value())));
+                gson.getAdapter(Map.class).write(writer, headers);
+            }
+        }
+
+        @Override
+        public Headers read(JsonReader reader) throws IOException {
+            Headers headers = null;
+            JsonToken peek = reader.peek();
+            if (JsonToken.NULL.equals(peek)) {
+                reader.nextNull();
+            } else {
+                Map<String, String> map = gson.getAdapter(Map.class).read(reader);
 
-        return (new Gson());
+                headers = new RecordHeaders();
+                for (Map.Entry<String, String> entry : map.entrySet()) {
+                    String key = entry.getKey();
+                    String value = entry.getValue();
+                    headers.add(key, value == null ? null : value.getBytes());
+                }
+            }
 
+            return headers;
+        }
     }
 
 }
diff --git a/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaConsumerHelper.java b/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaConsumerHelper.java
index a9ffd65e..27bb9e71 100644
--- a/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaConsumerHelper.java
+++ b/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaConsumerHelper.java
@@ -8,15 +8,20 @@ import com.google.gson.Gson;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
+
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
 import org.jsmart.zerocode.core.di.provider.GsonSerDeProvider;
 import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
 import org.jsmart.zerocode.core.kafka.consume.ConsumerLocalConfigs;
@@ -180,11 +185,19 @@ public class KafkaConsumerHelper {
 
             Object key = thisRecord.key();
             Object value = thisRecord.value();
-            LOGGER.info("\nRecord Key - {} , Record value - {}, Record partition - {}, Record offset - {}",
-                    key, value, thisRecord.partition(), thisRecord.offset());
+            Headers headers = thisRecord.headers();
+            LOGGER.info("\nRecord Key - {} , Record value - {}, Record partition - {}, Record offset - {}, Headers - {}",
+                    key, value, thisRecord.partition(), thisRecord.offset(), headers);
 
             JsonNode valueNode = objectMapper.readTree(value.toString());
-            ConsumerJsonRecord jsonRecord = new ConsumerJsonRecord(thisRecord.key(), null, valueNode);
+            Map<String, String> headersMap = null;
+            if (headers != null) {
+                headersMap = new HashMap<>();
+                for (Header header : headers) {
+                    headersMap.put(header.key(), new String(header.value()));
+                }
+            }
+            ConsumerJsonRecord jsonRecord = new ConsumerJsonRecord(thisRecord.key(), null, valueNode, headersMap);
             jsonRecords.add(jsonRecord);
         }
     }
diff --git a/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaProducerHelper.java b/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaProducerHelper.java
index 87f7d8d5..20ebe907 100644
--- a/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaProducerHelper.java
+++ b/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaProducerHelper.java
@@ -29,7 +29,7 @@ public class KafkaProducerHelper {
     private static final Gson gson = new GsonSerDeProvider().get();
     private static final ObjectMapper objectMapper = new ObjectMapperProvider().get();
 
-    public  static Producer<Long, String> createProducer(String bootStrapServers, String producerPropertyFile) {
+    public static Producer<Long, String> createProducer(String bootStrapServers, String producerPropertyFile) {
         try (InputStream propsIs = Resources.getResource(producerPropertyFile).openStream()) {
             Properties properties = new Properties();
             properties.load(propsIs);
@@ -57,21 +57,20 @@ public class KafkaProducerHelper {
                 recordToSend.partition(),
                 recordToSend.timestamp(),
                 recordToSend.key(),
-                recordToSend.value());
+                recordToSend.value(),
+                recordToSend.headers());
     }
 
-    public static ProducerRecord prepareJsonRecordToSend(String topicName, ProducerJsonRecord recordToSend) {
-
-        return new ProducerRecord(topicName,
-                //recordToSend.partition(),
-                //recordToSend.timestamp(),
+    public static ProducerRecord<Object, Object> prepareJsonRecordToSend(String topicName, ProducerJsonRecord<?> recordToSend) {
+        return ProducerRecordBuilder.from(topicName,
                 recordToSend.getKey(),
                 // --------------------------------------------
                 // It's a JSON as String. Nothing to worry !
                 // Kafka StringSerializer needs in this format.
                 // --------------------------------------------
-                recordToSend.getValue().toString()
-        );
+                recordToSend.getValue().toString())
+                .withHeaders(recordToSend.getHeaders())
+                .build();
     }
 
 
diff --git a/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/ProducerRecordBuilder.java b/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/ProducerRecordBuilder.java
new file mode 100644
index 00000000..1ee606de
--- /dev/null
+++ b/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/ProducerRecordBuilder.java
@@ -0,0 +1,39 @@
+package org.jsmart.zerocode.core.kafka.helper;
+
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+
+import java.util.Map;
+
+public class ProducerRecordBuilder {
+    private String topic;
+    private Object key;
+    private Object value;
+    private Headers headers;
+
+    private ProducerRecordBuilder() {}
+
+    public static ProducerRecordBuilder from(String topic, Object key, Object value) {
+        ProducerRecordBuilder producerRecordBuilder = new ProducerRecordBuilder();
+
+        producerRecordBuilder.topic = topic;
+        producerRecordBuilder.key = key;
+        producerRecordBuilder.value = value;
+
+        return producerRecordBuilder;
+    }
+
+    public ProducerRecordBuilder withHeaders(Map<String, String> headers) {
+        if (headers != null) {
+            this.headers = new RecordHeaders();
+            headers.forEach((hKey, hValue) -> this.headers.add(hKey, hValue.getBytes()));
+        }
+
+        return this;
+    }
+
+    public ProducerRecord<Object, Object> build() {
+        return new ProducerRecord<>(topic, null, null, key, value, headers);
+    }
+}
diff --git a/core/src/main/java/org/jsmart/zerocode/core/kafka/receive/message/ConsumerJsonRecord.java b/core/src/main/java/org/jsmart/zerocode/core/kafka/receive/message/ConsumerJsonRecord.java
index 4615f7e0..6739559f 100644
--- a/core/src/main/java/org/jsmart/zerocode/core/kafka/receive/message/ConsumerJsonRecord.java
+++ b/core/src/main/java/org/jsmart/zerocode/core/kafka/receive/message/ConsumerJsonRecord.java
@@ -4,19 +4,24 @@ import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.JsonNode;
 
+import java.util.Map;
+
 public class ConsumerJsonRecord<K> {
     private final K key;
     private final JsonNode jsonKey;
     private final JsonNode value;
+    private final Map<String, String> headers;
 
     @JsonCreator
     public ConsumerJsonRecord(
             @JsonProperty("key") K key,
             @JsonProperty("jsonKey") JsonNode jsonKey,
-            @JsonProperty("value") JsonNode value) {
+            @JsonProperty("value") JsonNode value,
+            @JsonProperty("headers") Map<String, String> headers) {
         this.key = key;
         this.jsonKey = jsonKey;
         this.value = value;
+        this.headers = headers;
     }
 
     public K getKey() {
@@ -31,6 +36,10 @@ public class ConsumerJsonRecord<K> {
         return value;
     }
 
+    public Map<String, String> getHeaders() {
+        return headers;
+    }
+
     @Override
     public String toString() {
         return "Record{" +
diff --git a/core/src/main/java/org/jsmart/zerocode/core/kafka/send/KafkaSender.java b/core/src/main/java/org/jsmart/zerocode/core/kafka/send/KafkaSender.java
index 83b1cdbe..318d6de4 100644
--- a/core/src/main/java/org/jsmart/zerocode/core/kafka/send/KafkaSender.java
+++ b/core/src/main/java/org/jsmart/zerocode/core/kafka/send/KafkaSender.java
@@ -61,6 +61,7 @@ public class KafkaSender {
             switch (recordType) {
                 case RAW:
                     rawRecords = gson.fromJson(requestJson, ProducerRawRecords.class);
+
                     String fileName = rawRecords.getFile();
                     if (fileName != null) {
                         File file = validateAndGetFile(fileName);
@@ -87,6 +88,7 @@ public class KafkaSender {
 
                 case JSON:
                     jsonRecords = objectMapper.readValue(requestJson, ProducerJsonRecords.class);
+
                     fileName = jsonRecords.getFile();
                     if (fileName != null) {
                         File file = validateAndGetFile(fileName);
@@ -112,7 +114,7 @@ public class KafkaSender {
             }
 
         } catch (Exception e) {
-            LOGGER.error("Error in sending record. Exception : " + e );
+            LOGGER.error("Error in sending record.", e);
             String failedStatus = objectMapper.writeValueAsString(new DeliveryDetails(FAILED, e.getMessage()));
             return prettyPrintJson(failedStatus);
 
@@ -131,7 +133,7 @@ public class KafkaSender {
         ProducerRecord qualifiedRecord = prepareRecordToSend(topicName, recordToSend);
 
         RecordMetadata metadata;
-        if (isAsync != null && isAsync == true) {
+        if (Boolean.TRUE.equals(isAsync)) {
             LOGGER.info("Asynchronous Producer sending record - {}", qualifiedRecord);
             metadata = (RecordMetadata) producer.send(qualifiedRecord, new ProducerAsyncCallback()).get();
         } else {
@@ -157,7 +159,7 @@ public class KafkaSender {
         ProducerRecord record = prepareJsonRecordToSend(topicName, recordToSend);
 
         RecordMetadata metadata;
-        if (isAsync != null && isAsync == true) {
+        if (Boolean.TRUE.equals(isAsync)) {
             LOGGER.info("Asynchronous - Producer sending JSON record - {}", record);
             metadata = (RecordMetadata) producer.send(record, new ProducerAsyncCallback()).get();
         } else {
diff --git a/core/src/main/java/org/jsmart/zerocode/core/kafka/send/message/ProducerJsonRecord.java b/core/src/main/java/org/jsmart/zerocode/core/kafka/send/message/ProducerJsonRecord.java
index 6cfa33dc..c4329676 100644
--- a/core/src/main/java/org/jsmart/zerocode/core/kafka/send/message/ProducerJsonRecord.java
+++ b/core/src/main/java/org/jsmart/zerocode/core/kafka/send/message/ProducerJsonRecord.java
@@ -4,20 +4,25 @@ import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.JsonNode;
 
+import java.util.Map;
+
 // TODO - add timestamp, partition key etc
 public class ProducerJsonRecord<K> {
     private final K key;
     private final JsonNode jsonKey;
     private final JsonNode value;
+    private final Map<String, String> headers;
 
     @JsonCreator
     public ProducerJsonRecord(
             @JsonProperty("key") K key,
             @JsonProperty("jsonKey") JsonNode jsonKey,
-            @JsonProperty("value") JsonNode value) {
+            @JsonProperty("value") JsonNode value,
+            @JsonProperty("headers") Map<String, String> headers) {
         this.key = key;
         this.jsonKey = jsonKey;
         this.value = value;
+        this.headers = headers;
     }
 
     public K getKey() {
@@ -32,12 +37,17 @@ public class ProducerJsonRecord<K> {
         return value;
     }
 
+    public Map<String, String> getHeaders() {
+        return headers;
+    }
+
     @Override
     public String toString() {
         return "Record{" +
                 "key='" + key + '\'' +
                 ", jsonKey=" + jsonKey +
                 ", value=" + value +
+                ", headers=" + headers +
                 '}';
     }
 }


Test patch:
diff --git a/core/src/test/java/org/jsmart/zerocode/core/kafka/helper/KafkaConsumerHelperTest.java b/core/src/test/java/org/jsmart/zerocode/core/kafka/helper/KafkaConsumerHelperTest.java
index 140006c3..bae73bb3 100644
--- a/core/src/test/java/org/jsmart/zerocode/core/kafka/helper/KafkaConsumerHelperTest.java
+++ b/core/src/test/java/org/jsmart/zerocode/core/kafka/helper/KafkaConsumerHelperTest.java
@@ -1,11 +1,25 @@
 package org.jsmart.zerocode.core.kafka.helper;
 
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.collect.Iterators;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.hamcrest.CoreMatchers;
 import org.jsmart.zerocode.core.kafka.consume.ConsumerLocalConfigs;
 import org.jsmart.zerocode.core.kafka.consume.ConsumerLocalConfigsWrap;
 import org.jsmart.zerocode.core.kafka.receive.ConsumerCommonConfigs;
+import org.jsmart.zerocode.core.kafka.receive.message.ConsumerJsonRecord;
+import org.junit.Assert;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.core.Is.is;
@@ -118,4 +132,25 @@ public class KafkaConsumerHelperTest {
         assertThat(consumerEffectiveConfigs.getCommitAsync(), is(true));
         assertThat(consumerEffectiveConfigs.getCommitSync(), is(false));
     }
+
+    @Test
+    public void should_read_json_with_headers_in_record() throws IOException {
+        // given
+        ConsumerRecord consumerRecord = Mockito.mock(ConsumerRecord.class);
+        Mockito.when(consumerRecord.key()).thenReturn("key");
+        Mockito.when(consumerRecord.value()).thenReturn("\"value\"");
+        Mockito.when(consumerRecord.headers())
+                .thenReturn(new RecordHeaders().add("headerKey", "headerValue".getBytes()));
+
+        // when
+        List<ConsumerJsonRecord> consumerJsonRecords = new ArrayList<>();
+        KafkaConsumerHelper.readJson(consumerJsonRecords, Iterators.forArray(consumerRecord));
+
+        // then
+        Assert.assertEquals(1, consumerJsonRecords.size());
+        ConsumerJsonRecord consumerJsonRecord = consumerJsonRecords.get(0);
+        Assert.assertEquals("key", consumerJsonRecord.getKey());
+        Assert.assertEquals("\"value\"", consumerJsonRecord.getValue().toString());
+        Assert.assertEquals(Collections.singletonMap("headerKey", "headerValue"), consumerJsonRecord.getHeaders());
+    }
 }
\ No newline at end of file
diff --git a/core/src/test/java/org/jsmart/zerocode/core/kafka/helper/ProducerRecordBuilderTest.java b/core/src/test/java/org/jsmart/zerocode/core/kafka/helper/ProducerRecordBuilderTest.java
new file mode 100644
index 00000000..11c7b5de
--- /dev/null
+++ b/core/src/test/java/org/jsmart/zerocode/core/kafka/helper/ProducerRecordBuilderTest.java
@@ -0,0 +1,43 @@
+package org.jsmart.zerocode.core.kafka.helper;
+
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.Header;
+import org.hamcrest.CoreMatchers;
+import org.junit.Test;
+
+import java.util.Collections;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
+
+public class ProducerRecordBuilderTest {
+
+    @Test
+    public void should_build_producer_record_without_headers() {
+        ProducerRecord<Object, Object> producerRecord = ProducerRecordBuilder
+                .from("topic", "key", "value")
+                .build();
+
+        assertMandatoryProperties(producerRecord);
+    }
+
+    @Test
+    public void should_build_producer_record_with_headers() {
+        ProducerRecord<Object, Object> producerRecord = ProducerRecordBuilder
+                .from("topic", "key", "value")
+                .withHeaders(Collections.singletonMap("headerKey", "headerValue"))
+                .build();
+
+        assertMandatoryProperties(producerRecord);
+        Header[] headers = producerRecord.headers().toArray();
+        assertEquals(1, headers.length);
+        assertEquals("headerKey", headers[0].key());
+        assertThat("headerValue".getBytes(), CoreMatchers.equalTo(headers[0].value()));
+    }
+
+    private void assertMandatoryProperties(ProducerRecord<Object, Object> producerRecord) {
+        assertEquals("topic", producerRecord.topic());
+        assertEquals("key", producerRecord.key());
+        assertEquals("value", producerRecord.value());
+    }
+}
diff --git a/core/src/test/java/org/jsmart/zerocode/core/kafka/receive/message/ConsumerJsonRecordTest.java b/core/src/test/java/org/jsmart/zerocode/core/kafka/receive/message/ConsumerJsonRecordTest.java
index 93567df2..77d61bfb 100644
--- a/core/src/test/java/org/jsmart/zerocode/core/kafka/receive/message/ConsumerJsonRecordTest.java
+++ b/core/src/test/java/org/jsmart/zerocode/core/kafka/receive/message/ConsumerJsonRecordTest.java
@@ -1,11 +1,16 @@
 package org.jsmart.zerocode.core.kafka.receive.message;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import org.hamcrest.CoreMatchers;
+import org.hamcrest.Matcher;
 import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
 import org.junit.Test;
 
 import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.core.Is.is;
@@ -16,26 +21,43 @@ public class ConsumerJsonRecordTest {
 
     @Test
     public void testSer() throws IOException {
+        // TODO: Use assert iso sysout
         String key = "key1";
         JsonNode value = objectMapper.readTree("\"val1\"");
 
-        ConsumerJsonRecord record = new ConsumerJsonRecord(key, null, value);
+        ConsumerJsonRecord<Object> record = new ConsumerJsonRecord<>(key, null, value, null);
         String json = objectMapper.writeValueAsString(record);
         System.out.println("1 json >> " + json);
 
 
         Integer key1 = 123;
-        record = new ConsumerJsonRecord(key1, null, value);
+        record = new ConsumerJsonRecord<>(key1, null, value, null);
         json = objectMapper.writeValueAsString(record);
         System.out.println("1 json >> " + json);
 
 
         Object key2 = 23.45;
-        record = new ConsumerJsonRecord(key2, null, value);
+        record = new ConsumerJsonRecord<>(key2, null, value, null);
         json = objectMapper.writeValueAsString(record);
         System.out.println("2 json >> " + json);
     }
 
+    @Test
+    public void should_serialize_a_record_with_headers() throws JsonProcessingException {
+        // given
+        JsonNode value = objectMapper.readTree("\"val\"");
+        Map<String, String> headers = new HashMap<>();
+        headers.put("hKey", "hValue");
+        headers.put("hKeyWithNullValue", null);
+        ConsumerJsonRecord<Object> record = new ConsumerJsonRecord<>("123", null, value, headers);
+
+        // when
+        String json = objectMapper.writeValueAsString(record);
+
+        // then
+        assertThat(json, CoreMatchers.equalTo("{\"key\":\"123\",\"jsonKey\":null,\"value\":\"val\",\"headers\":{\"hKey\":\"hValue\",\"hKeyWithNullValue\":null}}"));
+    }
+
     @Test
     public void testDeser_singleJsonRecord() throws IOException {
         String json = "{\n" +
diff --git a/core/src/test/java/org/jsmart/zerocode/core/kafka/send/message/ProducerJsonRecordsTest.java b/core/src/test/java/org/jsmart/zerocode/core/kafka/send/message/ProducerJsonRecordsTest.java
index bb26f38e..c18b31ed 100644
--- a/core/src/test/java/org/jsmart/zerocode/core/kafka/send/message/ProducerJsonRecordsTest.java
+++ b/core/src/test/java/org/jsmart/zerocode/core/kafka/send/message/ProducerJsonRecordsTest.java
@@ -1,11 +1,15 @@
 package org.jsmart.zerocode.core.kafka.send.message;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.jayway.jsonpath.JsonPath;
+import org.hamcrest.CoreMatchers;
 import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
 import org.junit.Test;
 
 import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.core.Is.is;
@@ -48,4 +52,22 @@ public class ProducerJsonRecordsTest {
         assertThat(producerRawRecords.getRecords().get(0).getKey(), is(101));
     }
 
+
+    @Test
+    public void testDeser_headers() throws JsonProcessingException {
+        final String json = "{\n" +
+                "                \"key\": 101,\n" +
+                "                \"value\": {\n" +
+                "                   \"name\" : \"Jey\"\n" +
+                "                },\n" +
+                "                \"headers\": {\n" +
+                "                    \"key\": \"value\"\n" +
+                "                }\n" +
+                "            }";
+
+        ProducerJsonRecord producerRawRecord = objectMapper.readValue(json, ProducerJsonRecord.class);
+        Map<String, String> expectedHeaders = new HashMap<>();
+        expectedHeaders.put("key", "value");
+        assertThat(producerRawRecord.getHeaders(), CoreMatchers.equalTo(expectedHeaders));
+    }
 }
\ No newline at end of file
diff --git a/core/src/test/java/org/jsmart/zerocode/core/kafka/send/message/ProducerRawRecordsTest.java b/core/src/test/java/org/jsmart/zerocode/core/kafka/send/message/ProducerRawRecordsTest.java
index ae56da9e..c2b57278 100644
--- a/core/src/test/java/org/jsmart/zerocode/core/kafka/send/message/ProducerRawRecordsTest.java
+++ b/core/src/test/java/org/jsmart/zerocode/core/kafka/send/message/ProducerRawRecordsTest.java
@@ -4,6 +4,8 @@ import com.google.gson.Gson;
 import com.google.gson.reflect.TypeToken;
 import com.jayway.jsonpath.JsonPath;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeaders;
 import org.jsmart.zerocode.core.di.provider.GsonSerDeProvider;
 import org.junit.Test;
 import org.skyscreamer.jsonassert.JSONAssert;
@@ -46,8 +48,17 @@ public class ProducerRawRecordsTest {
         producerRecord = new ProducerRecord("topic2", "key-123", "{\"name\": \"Nicola\"}");
         jsonBack = gson.toJson(producerRecord);
         assertThat(producerRecord.value(), is("{\"name\": \"Nicola\"}"));
-        JSONAssert.assertEquals("{\"topic\":\"topic2\",\"headers\":{\"headers\":[],\"isReadOnly\":false},\"key\":\"key-123\",\"value\":\"{\\\"name\\\": \\\"Nicola\\\"}\"}", jsonBack, LENIENT);
+        JSONAssert.assertEquals("{\"topic\":\"topic2\",\"key\":\"key-123\",\"value\":\"{\\\"name\\\": \\\"Nicola\\\"}\"}", jsonBack, LENIENT);
+    }
 
+    @Test
+    public void testDeser_headers() {
+        Headers headers = new RecordHeaders();
+        headers.add("headerKey1", "headerValue1".getBytes());
+        headers.add("headerKey2", "headerValue2".getBytes());
+        ProducerRecord producerRecord  = new ProducerRecord("topic2", null, "key-123", "Hello", headers);
+        String jsonBack = gson.toJson(producerRecord);
+        JSONAssert.assertEquals("{\"topic\":\"topic2\",\"headers\":{\"headerKey1\":\"headerValue1\",\"headerKey2\":\"headerValue2\"},\"key\":\"key-123\",\"value\":\"Hello\"}", jsonBack, LENIENT);
     }
 
     @Test
diff --git a/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/KafkaSuite.java b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/KafkaSuite.java
index 5c52580b..dd322bf1 100644
--- a/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/KafkaSuite.java
+++ b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/KafkaSuite.java
@@ -12,9 +12,11 @@ import org.jsmart.zerocode.integration.tests.kafka.produce.KafkaProduceAsyncTest
 import org.jsmart.zerocode.integration.tests.kafka.produce.KafkaProduceIntKeyTest;
 import org.jsmart.zerocode.integration.tests.kafka.produce.KafkaProduceJsonTest;
 import org.jsmart.zerocode.integration.tests.kafka.produce.KafkaProduceRawTest;
+import org.jsmart.zerocode.integration.tests.kafka.produce.KafkaProduceRawWithHeadersTest;
 import org.jsmart.zerocode.integration.tests.kafka.produce.KafkaProduceTest;
 import org.jsmart.zerocode.integration.tests.kafka.produce.KafkaProduceToPartitionTest;
 import org.jsmart.zerocode.integration.tests.kafka.produce.KafkaProduceTwoRecordsTest;
+import org.jsmart.zerocode.integration.tests.kafka.produce.KafkaProduceWithHeadersTest;
 import org.jsmart.zerocode.integration.tests.kafka.produce.KafkaProduceWithTimeStampTest;
 import org.jsmart.zerocode.integration.tests.kafka.produce.file.KafkaProduceAsyncFromFileRawTest;
 import org.jsmart.zerocode.integration.tests.kafka.produce.file.KafkaProduceSyncFromFileJsonTest;
@@ -34,6 +36,8 @@ import org.junit.runners.Suite;
         KafkaProduceTwoRecordsTest.class,
         KafkaProduceRawTest.class,
         KafkaProduceJsonTest.class,
+        KafkaProduceRawWithHeadersTest.class,
+        KafkaProduceWithHeadersTest.class,
         KafkaConsumeRawTest.class,
         KafkaConsumeJsonTest.class,
         KafkaProduceIntKeyTest.class,
diff --git a/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/produce/KafkaProduceRawWithHeadersTest.java b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/produce/KafkaProduceRawWithHeadersTest.java
new file mode 100644
index 00000000..4594e25c
--- /dev/null
+++ b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/produce/KafkaProduceRawWithHeadersTest.java
@@ -0,0 +1,18 @@
+package org.jsmart.zerocode.integration.tests.kafka.produce;
+
+import org.jsmart.zerocode.core.domain.Scenario;
+import org.jsmart.zerocode.core.domain.TargetEnv;
+import org.jsmart.zerocode.core.runner.ZeroCodeUnitRunner;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@TargetEnv("kafka_servers/kafka_test_server_double_key.properties")
+@RunWith(ZeroCodeUnitRunner.class)
+public class KafkaProduceRawWithHeadersTest {
+
+    @Test
+    @Scenario("kafka/produce/test_kafka_produce_raw_with_headers.json")
+    public void should_produce_raw_with_headers() {
+
+    }
+}
diff --git a/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/produce/KafkaProduceWithHeadersTest.java b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/produce/KafkaProduceWithHeadersTest.java
new file mode 100644
index 00000000..420d5923
--- /dev/null
+++ b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/produce/KafkaProduceWithHeadersTest.java
@@ -0,0 +1,18 @@
+package org.jsmart.zerocode.integration.tests.kafka.produce;
+
+import org.jsmart.zerocode.core.domain.Scenario;
+import org.jsmart.zerocode.core.domain.TargetEnv;
+import org.jsmart.zerocode.core.runner.ZeroCodeUnitRunner;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@TargetEnv("kafka_servers/kafka_test_server_int_key.properties")
+@RunWith(ZeroCodeUnitRunner.class)
+public class KafkaProduceWithHeadersTest {
+
+    @Test
+    @Scenario("kafka/produce/test_kafka_produce_json_with_headers.json")
+    public void should_produce_json_with_headers() {
+
+    }
+}
diff --git a/kafka-testing/src/test/resources/kafka/produce/test_kafka_produce_json_with_headers.json b/kafka-testing/src/test/resources/kafka/produce/test_kafka_produce_json_with_headers.json
new file mode 100644
index 00000000..8967e8fd
--- /dev/null
+++ b/kafka-testing/src/test/resources/kafka/produce/test_kafka_produce_json_with_headers.json
@@ -0,0 +1,56 @@
+{
+    "scenarioName": "Produce a message with headers to a kafka topic",
+    "steps": [
+        {
+            "name": "produce_to_kafka",
+            "url": "kafka-topic:demo-json-headers-topic",
+            "operation": "produce",
+            "request": {
+                "recordType": "JSON",
+                "records": [
+                    {
+                        "key": 101,
+                        "value": {
+                            "name": "Jey"
+                        },
+                        "headers": {
+                            "hKey": "something",
+                            "correlationId": "d85e88d2-3247-40a8-9c56-ec29004c45c9"
+                        }
+                    }
+                ]
+            },
+            "assertions": {
+                "status": "Ok",
+                "recordMetadata": "$NOT.NULL"
+            }
+        },
+        {
+            "name": "consume_from_kafka",
+            "url": "kafka-topic:demo-json-headers-topic",
+            "operation": "unload",
+            "request": {
+                "consumerLocalConfigs": {
+                    "recordType": "JSON",
+                    "commitSync": true,
+                    "maxNoOfRetryPollsOrTimeouts": 3
+                }
+            },
+            "assertions": {
+                "size": 1,
+                "records": [
+                    {
+                        "key": 101,
+                        "value": {
+                            "name": "Jey"
+                        },
+                        "headers": {
+                            "hKey": "something",
+                            "correlationId": "d85e88d2-3247-40a8-9c56-ec29004c45c9"
+                        }
+                    }
+                ]
+            }
+        }
+    ]
+}
diff --git a/kafka-testing/src/test/resources/kafka/produce/test_kafka_produce_raw_with_headers.json b/kafka-testing/src/test/resources/kafka/produce/test_kafka_produce_raw_with_headers.json
new file mode 100644
index 00000000..b10092c6
--- /dev/null
+++ b/kafka-testing/src/test/resources/kafka/produce/test_kafka_produce_raw_with_headers.json
@@ -0,0 +1,52 @@
+{
+    "scenarioName": "Produce a raw message with headers to a kafka topic",
+    "steps": [
+        {
+            "name": "produce_to_kafka",
+            "url": "kafka-topic:demo-raw-headers-topic",
+            "operation": "produce",
+            "request": {
+                "recordType": "RAW",
+                "records": [
+                    {
+                        "key": 101,
+                        "value": "Something",
+                        "headers": {
+                            "hKey": "something",
+                            "correlationId": "d85e88d2-3247-40a8-9c56-ec29004c45c9"
+                        }
+                    }
+                ]
+            },
+            "assertions": {
+                "status": "Ok",
+                "recordMetadata": "$NOT.NULL"
+            }
+        },
+        {
+            "name": "consume_from_kafka",
+            "url": "kafka-topic:demo-raw-headers-topic",
+            "operation": "consume",
+            "request": {
+                "consumerLocalConfigs": {
+                    "recordType": "RAW",
+                    "commitSync": true,
+                    "maxNoOfRetryPollsOrTimeouts": 3
+                }
+            },
+            "assertions": {
+                "size": "$GT.0",
+                "records": [
+                    {
+                        "key": 101,
+                        "value": "Something",
+                        "headers": {
+                            "hKey": "something",
+                            "correlationId": "d85e88d2-3247-40a8-9c56-ec29004c45c9"
+                        }
+                    }
+                ]
+            }
+        }
+    ]
+}