Repository: authorjapps/zerocode (java)

Issue:
[Kafka] Allow Json consume to natively read AVRO Message
# Context

For my organization project, i needed to produce and consume AVRO messages from  the Schema Registry so i develop a custom `KakfaClient` to produce the AVRO message.

Concerning the consumption, i detected after a debugging phase that the actual `KafkaReceiver` is almost able to treat an incoming AVRO message format  thanks to the `readJson` method

Here is the actual implementation 

```
            ConsumerJsonRecord jsonRecord = new ConsumerJsonRecord(thisRecord.key(), null, valueNode, headersMap);
```

Here is the code which allow parsing AVRO record 

```
            jsonRecords.add(new ConsumerJsonRecord(keyNode, null, valueNode, headersMap));

```
with `keyNode` 
```
            JsonNode keyNode = objectMapper.readTree(key.toString());

```

I can do a PR if this issue is accepted

Pull request:
ISSUE-488 # Allow Kafka receiver to natively read AVRO records
Close #488

# <Feature Title>

PR Branch
**_#ADD LINK TO THE PR BRANCH_**

## Motivation and Context

## 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
  * [x] 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/kafka/KafkaConstants.java b/core/src/main/java/org/jsmart/zerocode/core/kafka/KafkaConstants.java
index 7b88291e..7a8f9c54 100644
--- a/core/src/main/java/org/jsmart/zerocode/core/kafka/KafkaConstants.java
+++ b/core/src/main/java/org/jsmart/zerocode/core/kafka/KafkaConstants.java
@@ -9,7 +9,9 @@ public interface KafkaConstants {
     String RAW = "RAW";
 
     String JSON = "JSON";
-    
+
+    String AVRO = "AVRO";
+
     String PROTO = "PROTO";
 
     String RECORD_TYPE_JSON_PATH = "$.recordType";
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 357bc842..04fe8f48 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
@@ -4,6 +4,7 @@ import static java.lang.Integer.parseInt;
 import static java.lang.Long.parseLong;
 import static java.util.Optional.ofNullable;
 import static org.apache.commons.lang3.StringUtils.isEmpty;
+import static org.jsmart.zerocode.core.kafka.KafkaConstants.AVRO;
 import static org.jsmart.zerocode.core.kafka.KafkaConstants.DEFAULT_POLLING_TIME_MILLI_SEC;
 import static org.jsmart.zerocode.core.kafka.KafkaConstants.JSON;
 import static org.jsmart.zerocode.core.kafka.KafkaConstants.MAX_NO_OF_RETRY_POLLS_OR_TIME_OUTS;
@@ -220,11 +221,14 @@ public class KafkaConsumerHelper {
             Object key = thisRecord.key();
             Object valueObj = thisRecord.value();
             Headers headers = thisRecord.headers();
+            String keyStr =  thisRecord.key() != null ?  thisRecord.key().toString() : "";
             String valueStr = consumerLocalConfig != null && KafkaConstants.PROTO.equalsIgnoreCase(consumerLocalConfig.getRecordType()) ? convertProtobufToJson(thisRecord, consumerLocalConfig) : valueObj.toString();
             LOGGER.info("\nRecord Key - {} , Record value - {}, Record partition - {}, Record offset - {}, Headers - {}",
                     key, valueStr, thisRecord.partition(), thisRecord.offset(), headers);
 
+            JsonNode keyNode = objectMapper.readTree(keyStr);
             JsonNode valueNode = objectMapper.readTree(valueStr);
+
             Map<String, String> headersMap = null;
             if (headers != null) {
                 headersMap = new HashMap<>();
@@ -232,7 +236,7 @@ public class KafkaConsumerHelper {
                     headersMap.put(header.key(), new String(header.value()));
                 }
             }
-            ConsumerJsonRecord jsonRecord = new ConsumerJsonRecord(thisRecord.key(), null, valueNode, headersMap);
+            ConsumerJsonRecord jsonRecord = new ConsumerJsonRecord(keyNode, valueNode, headersMap);
             jsonRecords.add(jsonRecord);
         }
     }
@@ -276,7 +280,7 @@ public class KafkaConsumerHelper {
         } else if (testConfigs != null && RAW.equals(testConfigs.getRecordType())) {
             result = prettyPrintJson(gson.toJson(new ConsumerRawRecords(rawRecords)));
 
-        } else if (testConfigs != null && (JSON.equals(testConfigs.getRecordType()) || PROTO.equalsIgnoreCase(testConfigs.getRecordType()))) {
+        } else if (testConfigs != null && (JSON.equals(testConfigs.getRecordType()) || PROTO.equalsIgnoreCase(testConfigs.getRecordType()) || AVRO.equalsIgnoreCase(testConfigs.getRecordType()))) {
             result = prettyPrintJson(objectMapper.writeValueAsString(new ConsumerJsonRecords(jsonRecords)));
 
         } else {
diff --git a/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaFileRecordHelper.java b/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaFileRecordHelper.java
index 46d98844..bb79880d 100644
--- a/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaFileRecordHelper.java
+++ b/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaFileRecordHelper.java
@@ -16,6 +16,7 @@ import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.List;
 
+import static org.jsmart.zerocode.core.kafka.KafkaConstants.AVRO;
 import static org.jsmart.zerocode.core.kafka.KafkaConstants.JSON;
 import static org.jsmart.zerocode.core.kafka.KafkaConstants.RAW;
 import static org.jsmart.zerocode.core.kafka.KafkaConstants.PROTO;
@@ -37,6 +38,7 @@ public class KafkaFileRecordHelper {
                     dumpRawRecordsIfEnabled(consumeLocalTestProps.getFileDumpTo(), rawRecords);
                     break;
                 case PROTO:   
+                case AVRO:
                 case JSON:
                     dumpJsonRecordsIfEnabled(consumeLocalTestProps.getFileDumpTo(), jsonRecords);
                     break;
diff --git a/core/src/main/java/org/jsmart/zerocode/core/kafka/receive/KafkaReceiver.java b/core/src/main/java/org/jsmart/zerocode/core/kafka/receive/KafkaReceiver.java
index 9869189e..50f5db32 100644
--- a/core/src/main/java/org/jsmart/zerocode/core/kafka/receive/KafkaReceiver.java
+++ b/core/src/main/java/org/jsmart/zerocode/core/kafka/receive/KafkaReceiver.java
@@ -16,6 +16,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import static java.time.Duration.ofMillis;
+import static org.jsmart.zerocode.core.kafka.KafkaConstants.AVRO;
 import static org.jsmart.zerocode.core.kafka.KafkaConstants.JSON;
 import static org.jsmart.zerocode.core.kafka.KafkaConstants.RAW;
 import static org.jsmart.zerocode.core.kafka.KafkaConstants.PROTO;
@@ -108,7 +109,8 @@ public class KafkaReceiver {
             case RAW:
                 readRaw(rawRecords, recordIterator);
                 break;
-            case PROTO:    
+            case PROTO:
+            case AVRO:
             case JSON:
                 readJson(jsonRecords, recordIterator,effectiveLocal);
                 break;
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 6739559f..99a7fbdb 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
@@ -6,32 +6,25 @@ import com.fasterxml.jackson.databind.JsonNode;
 
 import java.util.Map;
 
-public class ConsumerJsonRecord<K> {
-    private final K key;
-    private final JsonNode jsonKey;
+public class ConsumerJsonRecord {
+    private final JsonNode key;
     private final JsonNode value;
     private final Map<String, String> headers;
 
     @JsonCreator
     public ConsumerJsonRecord(
-            @JsonProperty("key") K key,
-            @JsonProperty("jsonKey") JsonNode jsonKey,
+            @JsonProperty("key") JsonNode key,
             @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() {
+    public JsonNode getKey() {
         return key;
     }
 
-    public JsonNode getJsonKey() {
-        return jsonKey;
-    }
-
     public JsonNode getValue() {
         return value;
     }
@@ -44,7 +37,6 @@ public class ConsumerJsonRecord<K> {
     public String toString() {
         return "Record{" +
                 "key='" + key + '\'' +
-                ", jsonKey=" + jsonKey +
                 ", value=" + value +
                 '}';
     }
diff --git a/pom.xml b/pom.xml
index 50c81d9e..75e5df9b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -300,4 +300,4 @@
             </plugin>
         </plugins>
     </build>-->
-</project>
\ No newline at end of file
+</project>


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 fda81e60..f8471bab 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,5 +1,6 @@
 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.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -141,7 +142,7 @@ public class KafkaConsumerHelperTest {
     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.key()).thenReturn("\"key\"");
         Mockito.when(consumerRecord.value()).thenReturn("\"value\"");
         Mockito.when(consumerRecord.headers())
                 .thenReturn(new RecordHeaders().add("headerKey", "headerValue".getBytes()));
@@ -153,7 +154,9 @@ public class KafkaConsumerHelperTest {
         // then
         Assert.assertEquals(1, consumerJsonRecords.size());
         ConsumerJsonRecord consumerJsonRecord = consumerJsonRecords.get(0);
-        Assert.assertEquals("key", consumerJsonRecord.getKey());
+        Assert.assertTrue(consumerJsonRecord.getKey() instanceof JsonNode);
+        Assert.assertTrue(consumerJsonRecord.getValue() instanceof JsonNode);
+        Assert.assertEquals("\"key\"", consumerJsonRecord.getKey().toString());
         Assert.assertEquals("\"value\"", consumerJsonRecord.getValue().toString());
         Assert.assertEquals(Collections.singletonMap("headerKey", "headerValue"), consumerJsonRecord.getHeaders());
     }
@@ -222,4 +225,4 @@ public class KafkaConsumerHelperTest {
         // when
         ConsumerRecords records = initialPollWaitingForConsumerGroupJoin(consumer, consumerEffectiveConfigs);
     }
-}
\ No newline at end of file
+}
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 77d61bfb..aad106ca 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
@@ -22,22 +22,24 @@ public class ConsumerJsonRecordTest {
     @Test
     public void testSer() throws IOException {
         // TODO: Use assert iso sysout
-        String key = "key1";
+        JsonNode key = objectMapper.readTree("\"key1\"");
         JsonNode value = objectMapper.readTree("\"val1\"");
 
-        ConsumerJsonRecord<Object> record = new ConsumerJsonRecord<>(key, null, value, null);
+        ConsumerJsonRecord record = new ConsumerJsonRecord(key, value, null);
         String json = objectMapper.writeValueAsString(record);
         System.out.println("1 json >> " + json);
 
 
-        Integer key1 = 123;
-        record = new ConsumerJsonRecord<>(key1, null, value, null);
+        JsonNode key1 = objectMapper.readTree("123");
+
+        record = new ConsumerJsonRecord(key1, value, null);
         json = objectMapper.writeValueAsString(record);
         System.out.println("1 json >> " + json);
 
 
-        Object key2 = 23.45;
-        record = new ConsumerJsonRecord<>(key2, null, value, null);
+        JsonNode key2 = objectMapper.readTree("23.45");
+
+        record = new ConsumerJsonRecord(key2, value, null);
         json = objectMapper.writeValueAsString(record);
         System.out.println("2 json >> " + json);
     }
@@ -45,17 +47,18 @@ public class ConsumerJsonRecordTest {
     @Test
     public void should_serialize_a_record_with_headers() throws JsonProcessingException {
         // given
+        JsonNode key = objectMapper.readTree("123");
         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);
+        ConsumerJsonRecord record = new ConsumerJsonRecord(key, value, headers);
 
         // when
         String json = objectMapper.writeValueAsString(record);
 
         // then
-        assertThat(json, CoreMatchers.equalTo("{\"key\":\"123\",\"jsonKey\":null,\"value\":\"val\",\"headers\":{\"hKey\":\"hValue\",\"hKeyWithNullValue\":null}}"));
+        assertThat(json, CoreMatchers.equalTo("{\"key\":123,\"value\":\"val\",\"headers\":{\"hKey\":\"hValue\",\"hKeyWithNullValue\":null}}"));
     }
 
     @Test
@@ -69,4 +72,4 @@ public class ConsumerJsonRecordTest {
         ConsumerJsonRecord jsonRecord = objectMapper.readValue(json, ConsumerJsonRecord.class);
         assertThat(jsonRecord.getValue().toString(), is("{\"name\":\"Nicola\"}"));
     }
-}
\ No newline at end of file
+}
diff --git a/core/src/test/java/org/jsmart/zerocode/core/kafka/receive/message/ConsumerJsonRecordsTest.java b/core/src/test/java/org/jsmart/zerocode/core/kafka/receive/message/ConsumerJsonRecordsTest.java
index 0658e8a9..a0a7dbd5 100644
--- a/core/src/test/java/org/jsmart/zerocode/core/kafka/receive/message/ConsumerJsonRecordsTest.java
+++ b/core/src/test/java/org/jsmart/zerocode/core/kafka/receive/message/ConsumerJsonRecordsTest.java
@@ -1,6 +1,7 @@
 package org.jsmart.zerocode.core.kafka.receive.message;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import org.hamcrest.core.IsNot;
 import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
 import org.junit.Test;
 
@@ -8,6 +9,7 @@ import java.io.IOException;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.core.Is.is;
+import static org.hamcrest.core.IsNull.notNullValue;
 
 public class ConsumerJsonRecordsTest {
 
@@ -19,6 +21,9 @@ public class ConsumerJsonRecordsTest {
                 "                \"size\": 1,\n" +
                 "                \"records\": [\n" +
                 "                    {\n" +
+                "                       \"key\": {\n" +
+                "                            \"testkey\": \"keyvalue\"\n" +
+                "                        },\n" +
                 "                        \"value\": {\n" +
                 "                            \"name\": \"Jey\"\n" +
                 "                        }\n" +
@@ -28,6 +33,8 @@ public class ConsumerJsonRecordsTest {
 
         ConsumerJsonRecords jsonRecords = objectMapper.readValue(json, ConsumerJsonRecords.class);
         assertThat(jsonRecords.getRecords().get(0).getValue().toString(), is("{\"name\":\"Jey\"}"));
+        assertThat(jsonRecords.getRecords().get(0).getKey().get("testkey"), is(notNullValue()));
+        assertThat(jsonRecords.getRecords().get(0).getKey().get("testkey").textValue(), is("keyvalue"));
         assertThat(jsonRecords.getSize(), is(1));
     }
-}
\ No newline at end of file
+}
diff --git a/kafka-testing/pom.xml b/kafka-testing/pom.xml
index 28469dab..e7537903 100644
--- a/kafka-testing/pom.xml
+++ b/kafka-testing/pom.xml
@@ -26,6 +26,7 @@
 			<groupId>junit</groupId>
 			<artifactId>junit</artifactId>
 		</dependency>
+
 		<dependency>
 			<groupId>com.google.protobuf</groupId>
 			<artifactId>protobuf-java</artifactId>
@@ -38,7 +39,11 @@
 			<groupId>org.apache.kafka</groupId>
 			<artifactId>kafka-clients</artifactId>
 		</dependency>
-
+		<dependency>
+			<groupId>io.confluent</groupId>
+			<artifactId>kafka-avro-serializer</artifactId>
+			<version>5.1.0</version>
+		</dependency>
 		<dependency>
 			<groupId>com.github.os72</groupId>
 			<artifactId>protoc-jar</artifactId>
@@ -115,4 +120,11 @@
 			</plugin>
 		</plugins>
 	</build>
+	<repositories>
+		<repository>
+			<id>confluent</id>
+			<name>Confluent</name>
+			<url>https://packages.confluent.io/maven/</url>
+		</repository>
+	</repositories>
 </project>
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 781d6832..dad44351 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
@@ -7,6 +7,7 @@ import org.jsmart.zerocode.integration.tests.kafka.consume.KafkaConsumeRawTest;
 import org.jsmart.zerocode.integration.tests.kafka.consume.KafkaConsumeSeekOffsetTest;
 import org.jsmart.zerocode.integration.tests.kafka.consume.KafkaConsumeTest;
 import org.jsmart.zerocode.integration.tests.kafka.consume.KafkaConsumeXmlTest;
+import org.jsmart.zerocode.integration.tests.kafka.consume.KafkaProduceConsumeAvroTest;
 import org.jsmart.zerocode.integration.tests.kafka.consume.file.KafkaConsumeDumpToFileTest;
 import org.jsmart.zerocode.integration.tests.kafka.consume.negative.KafkaConsumeAvroNegativeTest;
 import org.jsmart.zerocode.integration.tests.kafka.produce.KafkaProduceAsyncTest;
@@ -47,6 +48,7 @@ import org.junit.runners.Suite;
         KafkaConsumeIntKeyTest.class,
         KafkaConsumeAvroTest.class,
         KafkaConsumeAvroNegativeTest.class,
+        KafkaProduceConsumeAvroTest.class,
         KafkaConsumeDumpToFileTest.class,
         KafkaProduceAsyncTest.class,
         KafkaProduceAsyncFromFileRawTest.class,
diff --git a/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/KafkaConsumeAvroTest.java b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/KafkaConsumeAvroTest.java
index 51cea216..c345821c 100644
--- a/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/KafkaConsumeAvroTest.java
+++ b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/KafkaConsumeAvroTest.java
@@ -3,16 +3,22 @@ package org.jsmart.zerocode.integration.tests.kafka.consume;
 import org.jsmart.zerocode.core.domain.Scenario;
 import org.jsmart.zerocode.core.domain.TargetEnv;
 import org.jsmart.zerocode.core.runner.ZeroCodeUnitRunner;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-
-@Ignore("Users Requested to ignore this until io.confluent:kafka-avro-serializer:5.1.0 becomes available at maven central." +
-        "But to see these tests Passing - Visit repo >> https://github.com/authorjapps/hello-kafka-stream-testing")
+// Uncommented after a contributer added the required dependencies to the POM.
+//@Ignore("Users Requested to ignore this until io.confluent:kafka-avro-serializer:5.1.0 becomes available at maven central." +
+//        "But to see these tests Passing - Visit repo >> https://github.com/authorjapps/hello-kafka-stream-testing")
 @TargetEnv("kafka_servers/kafka_test_server_avro.properties")
 @RunWith(ZeroCodeUnitRunner.class)
 public class KafkaConsumeAvroTest {
 
+    /**
+     * Note:
+     * None of these below tests uses key, hence whether it is Apache key-serializer
+     * or Confluent key-serializer makes no difference.
+     * The key-serializers were updated to Confluent key-serializer by the user to run the:
+     * ...zerocode/.../kafka/consume/KafkaProduceConsumeAvroTest.java (uses key and value in AVRO msg)
+     */
     @Test
     @Scenario("kafka/consume/test_kafka_consume_avro_msg_json.json")
     public void testKafkaConsume_avroJson() throws Exception {
diff --git a/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/KafkaProduceConsumeAvroTest.java b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/KafkaProduceConsumeAvroTest.java
new file mode 100644
index 00000000..77ab1ef5
--- /dev/null
+++ b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/KafkaProduceConsumeAvroTest.java
@@ -0,0 +1,17 @@
+package org.jsmart.zerocode.integration.tests.kafka.consume;
+
+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_avro.properties")
+@RunWith(ZeroCodeUnitRunner.class)
+public class KafkaProduceConsumeAvroTest {
+
+    @Test
+    @Scenario("kafka/produce-consume/test_kafka_produce_consume_avro_records.json")
+    public void testKafkaProduceConsume_avro_With_and_Without_Key() throws Exception {
+    }
+}
diff --git a/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/produce/KafkaProduceUniqueClientIdTest.java b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/produce/KafkaProduceUniqueClientIdTest.java
index 2a1754b3..4cb95e66 100644
--- a/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/produce/KafkaProduceUniqueClientIdTest.java
+++ b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/produce/KafkaProduceUniqueClientIdTest.java
@@ -6,7 +6,7 @@ import org.jsmart.zerocode.core.runner.ZeroCodeUnitRunner;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
-@TargetEnv("kafka_servers/kafka_test_server_unique.properties")
+@TargetEnv("kafka_servers/kafka_test_server.properties")
 @RunWith(ZeroCodeUnitRunner.class)
 public class KafkaProduceUniqueClientIdTest {
 
diff --git a/kafka-testing/src/test/resources/kafka/produce-consume/test_kafka_produce_consume_avro_records.json b/kafka-testing/src/test/resources/kafka/produce-consume/test_kafka_produce_consume_avro_records.json
new file mode 100755
index 00000000..489f9a2c
--- /dev/null
+++ b/kafka-testing/src/test/resources/kafka/produce-consume/test_kafka_produce_consume_avro_records.json
@@ -0,0 +1,167 @@
+{
+  "scenarioName": "Produce a JSON message to a kafka topic",
+  "steps": [
+    {
+      "name": "register_avro_schema_value_step",
+      "url": "http://localhost:8081/subjects/myavrorecord/versions",
+      "operation": "POST",
+      "request": {
+        "headers": {
+          "Content-Type": "application/vnd.schemaregistry.v1+json",
+          "Accept": "application/vnd.schemaregistry.v1+json, application/vnd.schemaregistry+json, application/json"
+        },
+        "body": {
+          "schema": "{\"type\":\"record\",\"name\":\"myavrorecord\",\"fields\":[{\"name\":\"avrof1\",\"type\":\"string\"}]}"
+        }
+      },
+      "assertions":
+      {
+        "status": 200,
+        "body": {
+          "id": "$IS.NOTNULL"
+        }
+      }
+    },
+    {
+      "name": "register_avro_schema_key_step",
+      "url": "http://localhost:8081/subjects/myavrorecordkey/versions",
+      "operation": "POST",
+      "request": {
+        "headers": {
+          "Content-Type": "application/vnd.schemaregistry.v1+json",
+          "Accept": "application/vnd.schemaregistry.v1+json, application/vnd.schemaregistry+json, application/json"
+        },
+        "body": {
+          "schema": "{\"type\":\"record\",\"name\":\"myavrorecordkey\",\"fields\":[{\"name\":\"key\",\"type\":\"string\"}]}"
+        }
+      },
+      "assertions":
+      {
+        "status": 200,
+        "body": {
+          "id": "$IS.NOTNULL"
+        }
+      }
+    },
+    {
+      "name": "produce_avro_msg_without_key_step",
+      "url": "/topics/demo-avro-12",
+      "operation": "POST",
+      "request": {
+        "headers": {
+          "Content-Type": "application/vnd.kafka.avro.v2+json",
+          "Accept": "application/vnd.kafka.v2+json"
+        },
+        "body": {
+          "value_schema_id": "${$.register_avro_schema_value_step.response.body.id}",
+          "records": [
+            {
+              "key": null,
+              "value": {
+                "avrof1": "it works"
+              }
+            }
+          ]
+        }
+      },
+    "assertions":
+    {
+      "status": 200,
+      "body": {
+        "offsets": [
+          {
+            "partition": "$NOT.NULL",
+            "offset": "$NOT.NULL"
+          }
+        ]
+      }
+    }
+    },
+    {
+      "name": "consume_avro_msg_without_key_as_avro",
+      "url": "kafka-topic:demo-avro-12",
+      "operation": "consume",
+      "request": {
+        "consumerLocalConfigs": {
+          "recordType": "AVRO",
+          "commitSync": true,
+          "showRecordsConsumed": true,
+          "maxNoOfRetryPollsOrTimeouts": 3
+        }
+      },
+      "assertions": {
+        "size": 1,
+        "records": [
+          {
+            "value": {
+              "avrof1": "it works"
+            }
+          }
+        ]
+      }
+    },
+    {
+      "name": "produce_avro_msg_with_key_step",
+      "url": "/topics/demo-avro-12",
+      "operation": "POST",
+      "request": {
+        "headers": {
+          "Content-Type": "application/vnd.kafka.avro.v2+json",
+          "Accept": "application/vnd.kafka.v2+json"
+        },
+        "body": {
+          "key_schema_id": "${$.register_avro_schema_key_step.response.body.id}",
+          "value_schema_id": "${$.register_avro_schema_value_step.response.body.id}",
+          "records": [
+            {
+              "key": {
+                "key": "key works"
+              },
+              "value": {
+                "avrof1": "it works"
+              }
+            }
+          ]
+        }
+      },
+      "assertions":
+      {
+        "status": 200,
+        "body": {
+          "offsets": [
+            {
+              "partition": "$NOT.NULL",
+              "offset": "$NOT.NULL"
+            }
+          ]
+        }
+      }
+    },
+    {
+      "name": "consume_avro_msg_with_key_as_avro",
+      "url": "kafka-topic:demo-avro-12",
+      "operation": "consume",
+      "request": {
+        "consumerLocalConfigs": {
+          "recordType": "AVRO",
+          "commitSync": true,
+          "showRecordsConsumed": true,
+          "maxNoOfRetryPollsOrTimeouts": 3
+        }
+      },
+      "assertions": {
+        "size": 1,
+        "records": [
+          {
+            "key": {
+              "key": "key works"
+            },
+            "value": {
+              "avrof1": "it works"
+            }
+          }
+        ]
+      }
+    }
+  ]
+}
diff --git a/kafka-testing/src/test/resources/kafka_servers/kafka_consumer_avro.properties b/kafka-testing/src/test/resources/kafka_servers/kafka_consumer_avro.properties
index f6bbb2ce..6cff8788 100755
--- a/kafka-testing/src/test/resources/kafka_servers/kafka_consumer_avro.properties
+++ b/kafka-testing/src/test/resources/kafka_servers/kafka_consumer_avro.properties
@@ -2,8 +2,9 @@
 #          kafka consumer properties
 # =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
 group.id=consumerGroup11
-key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
+key.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer
 value.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer
+
 schema.registry.url=http://localhost:8081
 max.poll.records=2
 enable.auto.commit=false
diff --git a/kafka-testing/src/test/resources/kafka_servers/kafka_producer_avro.properties b/kafka-testing/src/test/resources/kafka_servers/kafka_producer_avro.properties
index f379ce04..2d92b834 100755
--- a/kafka-testing/src/test/resources/kafka_servers/kafka_producer_avro.properties
+++ b/kafka-testing/src/test/resources/kafka_servers/kafka_producer_avro.properties
@@ -2,8 +2,7 @@
 #          kafka producer properties
 # =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
 client.id=zerocode-producer
-key.serializer=org.apache.kafka.common.serialization.StringSerializer
-#value.serializer=org.apache.kafka.common.serialization.StringSerializer
+key.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer
 value.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer
 schema.registry.url=http://localhost:8081