Repository: authorjapps/zerocode
Pull request #552 (2022-11-04 16:50:36)

Kafka consumer cache
# <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

* [ ] 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/kafka/consume/ConsumerLocalConfigs.java b/core/src/main/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigs.java
index ac9b3812..86599251 100644
--- a/core/src/main/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigs.java
+++ b/core/src/main/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigs.java
@@ -21,6 +21,7 @@ public class ConsumerLocalConfigs {
     private final Long pollingTime;
     private final String seek;
     private final String protoClassType;
+    private final Boolean cacheByTopic;
     private final String filterByJsonPath;
 
     @JsonCreator
@@ -33,6 +34,7 @@ public class ConsumerLocalConfigs {
             @JsonProperty("showRecordsConsumed") Boolean showRecordsConsumed,
             @JsonProperty("maxNoOfRetryPollsOrTimeouts") Integer maxNoOfRetryPollsOrTimeouts,
             @JsonProperty("pollingTime") Long pollingTime,
+            @JsonProperty("cacheByTopic") Boolean cacheByTopic,
             @JsonProperty("filterByJsonPath") String filterByJsonPath,
             @JsonProperty("seek") String seek) {
         this.recordType = recordType;
@@ -43,6 +45,7 @@ public class ConsumerLocalConfigs {
         this.showRecordsConsumed = showRecordsConsumed;
         this.maxNoOfRetryPollsOrTimeouts = maxNoOfRetryPollsOrTimeouts;
         this.pollingTime = pollingTime;
+        this.cacheByTopic = cacheByTopic;
         this.filterByJsonPath = filterByJsonPath;
         this.seek = seek;
     }
@@ -56,6 +59,7 @@ public class ConsumerLocalConfigs {
             Boolean showRecordsConsumed,
             Integer maxNoOfRetryPollsOrTimeouts,
             Long pollingTime,
+            Boolean cacheByTopic,
             String filterByJsonPath,
             String seek) {
 		this(recordType, null,
@@ -65,6 +69,7 @@ public class ConsumerLocalConfigs {
                 showRecordsConsumed,
                 maxNoOfRetryPollsOrTimeouts,
 				pollingTime,
+                cacheByTopic,
                 filterByJsonPath,
                 seek);
     }
@@ -102,6 +107,10 @@ public class ConsumerLocalConfigs {
         return pollingTime;
     }
 
+    public Boolean getCacheByTopic() {
+        return cacheByTopic;
+    }
+
     public String getFilterByJsonPath() {
         return filterByJsonPath;
     }
@@ -129,13 +138,14 @@ public class ConsumerLocalConfigs {
                 Objects.equals(maxNoOfRetryPollsOrTimeouts, that.maxNoOfRetryPollsOrTimeouts) &&
                 Objects.equals(pollingTime, that.pollingTime) &&
                 Objects.equals(filterByJsonPath, that.filterByJsonPath) &&
+                Objects.equals(cacheByTopic, that.cacheByTopic) &&
                 Objects.equals(seek, that.seek);
     }
 
     @Override
     public int hashCode() {
 
-        return Objects.hash(recordType, fileDumpTo, commitAsync, commitSync, showRecordsConsumed, maxNoOfRetryPollsOrTimeouts, pollingTime, filterByJsonPath, seek);
+        return Objects.hash(recordType, fileDumpTo, commitAsync, commitSync, showRecordsConsumed, maxNoOfRetryPollsOrTimeouts, pollingTime,cacheByTopic, filterByJsonPath, seek);
     }
 
     @Override
@@ -149,6 +159,7 @@ public class ConsumerLocalConfigs {
                 ", showRecordsConsumed=" + showRecordsConsumed +
                 ", maxNoOfRetryPollsOrTimeouts=" + maxNoOfRetryPollsOrTimeouts +
                 ", pollingTime=" + pollingTime +
+                ", cacheByTopic=" + cacheByTopic +
                 ", filterByJsonPath=" + filterByJsonPath +
                 ", seek=" + seek +
                 '}';
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 9632b7a8..5f56e85f 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
@@ -19,7 +19,6 @@ import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.time.Duration;
 import java.time.temporal.ChronoUnit;
-import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -29,7 +28,6 @@ import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
 
-import com.fasterxml.jackson.core.type.TypeReference;
 import com.jayway.jsonpath.JsonPath;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -64,8 +62,15 @@ public class KafkaConsumerHelper {
     private static final Logger LOGGER = LoggerFactory.getLogger(KafkaConsumerHelper.class);
     private static final Gson gson = new GsonSerDeProvider().get();
     private static final ObjectMapper objectMapper = new ObjectMapperProvider().get();
+    public static final String CONSUMER = "CONSUMER";
+    public static Map<String, Consumer> consumerCacheByTopicMap = new HashMap<>();
+
+    public static Consumer createConsumer(String bootStrapServers, String consumerPropertyFile, String topic, Boolean consumerToBeCached) {
+        Consumer sameConsumer = getCachedConsumer(topic, consumerToBeCached);
+        if (sameConsumer != null) {
+            return sameConsumer;
+        }
 
-    public static Consumer createConsumer(String bootStrapServers, String consumerPropertyFile, String topic) {
         try (InputStream propsIs = Resources.getResource(consumerPropertyFile).openStream()) {
             Properties properties = new Properties();
             properties.load(propsIs);
@@ -76,6 +81,27 @@ public class KafkaConsumerHelper {
             final Consumer consumer = new KafkaConsumer(properties);
             consumer.subscribe(Collections.singletonList(topic));
 
+            if(consumerToBeCached == true){
+                consumerCacheByTopicMap.forEach((xTopic, xConsumer) -> {
+                    if(!xTopic.equals(topic)){
+                        // close the earlier consumer if in the same group for safety.
+                        // (even if not in the same group, closing it anyway will not do any harm)
+                        // Otherwise rebalance will fail while rejoining/joining the same group for a new consumer
+                        // i.e. because old consumer(xConsumer) is still consuming,
+                        // and has not let GC know that it has stopped consuming or not sent any LeaveGroup request.
+                        // If you have a single(0) partition topic in your Kafka Broker, xConsumer is still holding it,
+                        // i.e. not yet unassigned.
+                        // Note- It works fine and not required to close() if the new consumer joining the same Group for the same topic.
+                        xConsumer.close();
+                    }
+                });
+                // Remove the earlier topic-consumer from the cache.
+                // Recreate will happen above anyway if not found in cache via "new KafkaConsumer(properties)".
+                consumerCacheByTopicMap.entrySet().removeIf(xTopic -> !xTopic.equals(topic));
+
+                consumerCacheByTopicMap.put(topic, consumer);
+            }
+
             return consumer;
 
         } catch (IOException e) {
@@ -87,9 +113,12 @@ public class KafkaConsumerHelper {
 
             for (int run = 0; run < 50; run++) {
                 if (!consumer.assignment().isEmpty()) {
+                    LOGGER.info("==> WaitingForConsumerGroupJoin - Partition now assigned. No records yet consumed");
                     return new ConsumerRecords(new HashMap());
                 }
+                LOGGER.info("==> WaitingForConsumerGroupJoin - Partition not assigned. Polling once");
                 ConsumerRecords records = consumer.poll(Duration.of(getPollTime(effectiveLocalConfigs), ChronoUnit.MILLIS));
+                LOGGER.info("==> WaitingForConsumerGroupJoin - polled records length={}", records.count());
                 if (!records.isEmpty()) {
                     return records;
                 }
@@ -132,6 +161,7 @@ public class KafkaConsumerHelper {
                     consumerCommon.getShowRecordsConsumed(),
                     consumerCommon.getMaxNoOfRetryPollsOrTimeouts(),
                     consumerCommon.getPollingTime(),
+                    consumerCommon.getCacheByTopic(),
                     consumerCommon.getFilterByJsonPath(),
                     consumerCommon.getSeek());
         }
@@ -161,6 +191,10 @@ public class KafkaConsumerHelper {
         // Handle pollingTime
         String effectiveSeek = ofNullable(consumerLocal.getSeek()).orElse(consumerCommon.getSeek());
 
+        // Handle consumerCache by topic
+        Boolean effectiveConsumerCacheByTopic = ofNullable(consumerLocal.getCacheByTopic())
+                .orElse(consumerCommon.getCacheByTopic());
+
         // Handle commitSync and commitAsync -START
         Boolean effectiveCommitSync;
         Boolean effectiveCommitAsync;
@@ -186,6 +220,7 @@ public class KafkaConsumerHelper {
                 effectiveShowRecordsConsumed,
                 effectiveMaxNoOfRetryPollsOrTimeouts,
                 effectivePollingTime,
+                effectiveConsumerCacheByTopic,
                 filterByJsonPath,
                 effectiveSeek);
     }
@@ -382,4 +417,12 @@ public class KafkaConsumerHelper {
             }
         }
     }
+
+    private static Consumer getCachedConsumer(String topic, Boolean consumerToBeCached) {
+        if(consumerToBeCached){
+            return consumerCacheByTopicMap.get(topic);
+        }
+        return null;
+    }
+
 }
diff --git a/core/src/main/java/org/jsmart/zerocode/core/kafka/receive/ConsumerCommonConfigs.java b/core/src/main/java/org/jsmart/zerocode/core/kafka/receive/ConsumerCommonConfigs.java
index 2c612abe..2b6640d4 100644
--- a/core/src/main/java/org/jsmart/zerocode/core/kafka/receive/ConsumerCommonConfigs.java
+++ b/core/src/main/java/org/jsmart/zerocode/core/kafka/receive/ConsumerCommonConfigs.java
@@ -39,6 +39,12 @@ public class ConsumerCommonConfigs {
     @Named("consumer.pollingTime")
     private Long pollingTime;
 
+    @Inject(optional = true)
+    @Named("consumer.cacheByTopic")
+    private Boolean cacheByTopic = false;
+
+    // TODO - Delete this config from common configs.
+    // This is not needed as global settings(double check)
     @Inject(optional = true)
     @Named("consumer.filterByJsonPath")
     private String filterByJsonPath;
@@ -59,6 +65,7 @@ public class ConsumerCommonConfigs {
                                  Boolean showRecordsConsumed,
                                  Integer maxNoOfRetryPollsOrTimeouts,
                                  Long pollingTime,
+                                 Boolean cacheByTopic,
                                  String filterByJsonPath,
                                  String seek
 
@@ -71,6 +78,7 @@ public class ConsumerCommonConfigs {
         this.showRecordsConsumed = showRecordsConsumed;
         this.maxNoOfRetryPollsOrTimeouts = maxNoOfRetryPollsOrTimeouts;
         this.pollingTime = pollingTime;
+        this.cacheByTopic = cacheByTopic;
         this.filterByJsonPath = filterByJsonPath;
         this.seek = seek;
     }
@@ -82,6 +90,7 @@ public class ConsumerCommonConfigs {
             Boolean showRecordsConsumed,
             Integer maxNoOfRetryPollsOrTimeouts,
             Long pollingTime,
+            Boolean cacheByTopic,
             String filterByJsonPath,
             String seek
 
@@ -94,6 +103,7 @@ public class ConsumerCommonConfigs {
                 showRecordsConsumed,
                 maxNoOfRetryPollsOrTimeouts,
 				pollingTime,
+                cacheByTopic,
                 filterByJsonPath,
                 seek);
 	}
@@ -134,6 +144,10 @@ public class ConsumerCommonConfigs {
 	}
 
 
+    public Boolean getCacheByTopic() {
+        return cacheByTopic;
+    }
+
     public String getFilterByJsonPath() {
         return filterByJsonPath;
     }
@@ -149,6 +163,7 @@ public class ConsumerCommonConfigs {
                 ", showRecordsConsumed=" + showRecordsConsumed +
                 ", maxNoOfRetryPollsOrTimeouts=" + maxNoOfRetryPollsOrTimeouts +
                 ", pollingTime=" + pollingTime +
+                ", cacheByTopic=" + cacheByTopic +
                 ", filterByJsonPath=" + filterByJsonPath +
                 ", seek=" + seek +
                 '}';
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 50f5db32..e888c8e8 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
@@ -53,7 +53,9 @@ public class KafkaReceiver {
 
         LOGGER.info("\n### Kafka Consumer Effective configs:{}\n", effectiveLocal);
 
-        Consumer consumer = createConsumer(kafkaServers, consumerPropertyFile, topicName);
+        Consumer consumer = createConsumer(kafkaServers,
+                consumerPropertyFile, topicName,
+                consumerCommonConfigs.getCacheByTopic());
 
         final List<ConsumerRecord> rawRecords = new ArrayList<>();
         final List<ConsumerJsonRecord> jsonRecords = new ArrayList<>();
@@ -92,7 +94,9 @@ public class KafkaReceiver {
 
         }
 
-        consumer.close();
+        if(!consumerCommonConfigs.getCacheByTopic()){
+            consumer.close();
+        }
 
         handleRecordsDump(effectiveLocal, rawRecords, jsonRecords);
 
diff --git a/core/src/main/java/org/jsmart/zerocode/core/runner/ZeroCodePackageRunner.java b/core/src/main/java/org/jsmart/zerocode/core/runner/ZeroCodePackageRunner.java
index 81ea2f3b..d1b82929 100644
--- a/core/src/main/java/org/jsmart/zerocode/core/runner/ZeroCodePackageRunner.java
+++ b/core/src/main/java/org/jsmart/zerocode/core/runner/ZeroCodePackageRunner.java
@@ -19,7 +19,7 @@ import org.jsmart.zerocode.core.domain.TargetEnv;
 import org.jsmart.zerocode.core.domain.TestPackageRoot;
 import org.jsmart.zerocode.core.domain.UseHttpClient;
 import org.jsmart.zerocode.core.domain.UseKafkaClient;
-import org.jsmart.zerocode.core.engine.listener.ZeroCodeTestReportListener;
+import org.jsmart.zerocode.core.engine.listener.TestUtilityListener;
 import org.jsmart.zerocode.core.httpclient.BasicHttpClient;
 import org.jsmart.zerocode.core.httpclient.ssl.SslTrustHttpClient;
 import org.jsmart.zerocode.core.kafka.client.BasicKafkaClient;
@@ -139,7 +139,7 @@ public class ZeroCodePackageRunner extends ParentRunner<ScenarioSpec> {
 
     @Override
     public void run(RunNotifier notifier) {
-        RunListener reportListener = createReportListener();
+        RunListener reportListener = createTestUtilityListener();
         notifier.addListener(reportListener);
 
         LOGGER.info("System property " + ZEROCODE_JUNIT + "=" + getProperty(ZEROCODE_JUNIT));
@@ -152,8 +152,8 @@ public class ZeroCodePackageRunner extends ParentRunner<ScenarioSpec> {
         handleNoRunListenerReport(reportListener);
     }
 
-    protected RunListener createReportListener() {
-        return getMainModuleInjector().getInstance(ZeroCodeTestReportListener.class);
+    protected RunListener createTestUtilityListener() {
+        return getMainModuleInjector().getInstance(TestUtilityListener.class);
     }
 
 
diff --git a/core/src/main/java/org/jsmart/zerocode/core/runner/ZeroCodeUnitRunner.java b/core/src/main/java/org/jsmart/zerocode/core/runner/ZeroCodeUnitRunner.java
index 2fe2e4f0..56964a76 100644
--- a/core/src/main/java/org/jsmart/zerocode/core/runner/ZeroCodeUnitRunner.java
+++ b/core/src/main/java/org/jsmart/zerocode/core/runner/ZeroCodeUnitRunner.java
@@ -20,7 +20,7 @@ import org.jsmart.zerocode.core.domain.UseHttpClient;
 import org.jsmart.zerocode.core.domain.UseKafkaClient;
 import org.jsmart.zerocode.core.domain.builders.ZeroCodeExecReportBuilder;
 import org.jsmart.zerocode.core.domain.builders.ZeroCodeIoWriteBuilder;
-import org.jsmart.zerocode.core.engine.listener.ZeroCodeTestReportListener;
+import org.jsmart.zerocode.core.engine.listener.TestUtilityListener;
 import org.jsmart.zerocode.core.httpclient.BasicHttpClient;
 import org.jsmart.zerocode.core.httpclient.ssl.SslTrustHttpClient;
 import org.jsmart.zerocode.core.kafka.client.BasicKafkaClient;
@@ -97,7 +97,7 @@ public class ZeroCodeUnitRunner extends BlockJUnit4ClassRunner {
 
     @Override
     public void run(RunNotifier notifier) {
-        RunListener reportListener = createReportListener();
+        RunListener reportListener = createTestUtilityListener();
 
         LOGGER.info("System property " + ZEROCODE_JUNIT + "=" + getProperty(ZEROCODE_JUNIT));
         if (!CHARTS_AND_CSV.equals(getProperty(ZEROCODE_JUNIT))) {
@@ -193,8 +193,8 @@ public class ZeroCodeUnitRunner extends BlockJUnit4ClassRunner {
      * End User experience can be enhanced via this
      * @return An instance of the Junit RunListener
      */
-    protected RunListener createReportListener() {
-        return getMainModuleInjector().getInstance(ZeroCodeTestReportListener.class);
+    protected RunListener createTestUtilityListener() {
+        return getMainModuleInjector().getInstance(TestUtilityListener.class);
     }
 
     protected SmartUtils getInjectedSmartUtilsClass() {
diff --git a/docker/compose/shutdown.sh b/docker/compose/shutdown.sh
new file mode 100755
index 00000000..099b0df3
--- /dev/null
+++ b/docker/compose/shutdown.sh
@@ -0,0 +1,6 @@
+#!/bin/bash
+echo "shutting down confluent kafka..."
+SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
+docker-compose -f $SCRIPT_DIR/kafka-schema-registry.yml kill
+docker-compose -f $SCRIPT_DIR/kafka-schema-registry.yml rm -f
+echo "Done."
diff --git a/pom.xml b/pom.xml
index 0396f61c..310d5448 100644
--- a/pom.xml
+++ b/pom.xml
@@ -82,7 +82,7 @@
         <httpclient.version>4.5</httpclient.version>
         <h2.db.version>1.4.191</h2.db.version>
         <extentreports.version>4.0.9</extentreports.version>
-        <version.kafka-clients>2.1.0</version.kafka-clients>
+        <version.kafka-clients>3.3.1</version.kafka-clients>
         <version.gson>2.6.2</version.gson>
         <version.univocity-parsers>2.8.2</version.univocity-parsers>
         <!-- Maven plugins, build properties  -->


Test patch:
diff --git a/core/src/main/java/org/jsmart/zerocode/core/engine/listener/ZeroCodeTestReportListener.java b/core/src/main/java/org/jsmart/zerocode/core/engine/listener/TestUtilityListener.java
similarity index 80%
rename from core/src/main/java/org/jsmart/zerocode/core/engine/listener/ZeroCodeTestReportListener.java
rename to core/src/main/java/org/jsmart/zerocode/core/engine/listener/TestUtilityListener.java
index d729ba18..64feb8cc 100644
--- a/core/src/main/java/org/jsmart/zerocode/core/engine/listener/ZeroCodeTestReportListener.java
+++ b/core/src/main/java/org/jsmart/zerocode/core/engine/listener/TestUtilityListener.java
@@ -9,15 +9,15 @@ import org.junit.runner.notification.RunListener;
 
 import static org.slf4j.LoggerFactory.getLogger;
 
-public class ZeroCodeTestReportListener extends RunListener {
-    private static final org.slf4j.Logger LOGGER = getLogger(ZeroCodeTestReportListener.class);
+public class TestUtilityListener extends RunListener {
+    private static final org.slf4j.Logger LOGGER = getLogger(TestUtilityListener.class);
 
     private final ObjectMapper mapper;
 
     private final ZeroCodeReportGenerator reportGenerator;
 
     @Inject
-    public ZeroCodeTestReportListener(ObjectMapper mapper, ZeroCodeReportGenerator injectedReportGenerator) {
+    public TestUtilityListener(ObjectMapper mapper, ZeroCodeReportGenerator injectedReportGenerator) {
         this.mapper = mapper;
         this.reportGenerator = injectedReportGenerator;
     }
@@ -41,7 +41,7 @@ public class ZeroCodeTestReportListener extends RunListener {
     }
 
     private void printTestCompleted() {
-        LOGGER.info("#ZeroCode: Generating test reports...");
+        LOGGER.info("Generating test-statistics reports...");
         LOGGER.debug("#ZeroCode: Test run completed for this runner. Generating test reports... " +
                 "\n* For more examples, visit https://github.com/authorjapps/zerocode/wiki");
     }
@@ -60,8 +60,8 @@ public class ZeroCodeTestReportListener extends RunListener {
         reportGenerator.generateCsvReport();
 
         /**
-         * Not compatible with open source license i.e. why not activated But if it has to be used inside intranet,
-         * then a single Developer's license should do. But visit www.highcharts.com for details.
+         * Not compatible with open source license i.e. why not activated. But if it has to be used inside intranet,
+         * then a single Developer's license should do. Anyway visit www.highcharts.com for details.
 
          * https://shop.highsoft.com/faq
          * If I am using the Software on a commercial company´s intranet, does it require a license?
diff --git a/core/src/test/java/org/jsmart/zerocode/core/kafka/DeliveryDetailsTest.java b/core/src/test/java/org/jsmart/zerocode/core/kafka/DeliveryDetailsTest.java
index 89f9456b..a646c5de 100644
--- a/core/src/test/java/org/jsmart/zerocode/core/kafka/DeliveryDetailsTest.java
+++ b/core/src/test/java/org/jsmart/zerocode/core/kafka/DeliveryDetailsTest.java
@@ -61,22 +61,20 @@ public class DeliveryDetailsTest {
         jsonMsg = gson.toJson(deliveryDetails);
 
         JSONAssert.assertEquals("{\n" +
-                "    \"status\": \"Ok\",\n" +
-                "    \"recordMetadata\": {\n" +
-                "        \"offset\": 2,\n" +
-                "        \"timestamp\": 1546008192846,\n" +
-                "        \"serializedKeySize\": 1,\n" +
-                "        \"serializedValueSize\": 45,\n" +
-                "        \"topicPartition\": {\n" +
-                "            \"hash\": 0,\n" +
-                "            \"partition\": 0,\n" +
-                "            \"topic\": \"test-topic\"\n" +
-                "        },\n" +
-                "        \"checksum\": 100\n" +
-                "    }\n" +
-                "}",
+                        "    \"status\": \"Ok\",\n" +
+                        "    \"recordMetadata\": {\n" +
+                        "        \"offset\": 2,\n" +
+                        "        \"timestamp\": 1546008192846,\n" +
+                        "        \"serializedKeySize\": 1,\n" +
+                        "        \"serializedValueSize\": 45,\n" +
+                        "        \"topicPartition\": {\n" +
+                        "            \"hash\": 0,\n" +
+                        "            \"partition\": 0,\n" +
+                        "            \"topic\": \"test-topic\"\n" +
+                        "        }\n" +
+                        "    }\n" +
+                        "}",
                 jsonMsg, LENIENT);
 
     }
-
-}
+}
\ No newline at end of file
diff --git a/core/src/test/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigsWrapTest.java b/core/src/test/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigsWrapTest.java
index 0e4eeb90..4a9de75d 100644
--- a/core/src/test/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigsWrapTest.java
+++ b/core/src/test/java/org/jsmart/zerocode/core/kafka/consume/ConsumerLocalConfigsWrapTest.java
@@ -25,6 +25,7 @@ public class ConsumerLocalConfigsWrapTest {
                         true,
                         3,
                         50L,
+                        false,
                         "$.JSON.Path",
                         "1,0,test-topic"));
         ObjectMapper objectMapper = new ObjectMapperProvider().get();
@@ -55,6 +56,7 @@ public class ConsumerLocalConfigsWrapTest {
                         false,
                         3,
                         null,
+                        false,
                         null,
                         "1,0,test-topic"));
 
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 2b44da5c..e1b2eefd 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,228 +1,229 @@
-//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;
-//import org.apache.kafka.clients.consumer.ConsumerRecords;
-//import org.apache.kafka.common.TopicPartition;
-//import org.apache.kafka.common.header.internals.RecordHeaders;
-//import org.jsmart.zerocode.core.kafka.KafkaConstants;
-//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.Mockito;
-//
-//import java.io.IOException;
-//import java.time.Duration;
-//import java.util.ArrayList;
-//import java.util.Collections;
-//import java.util.HashSet;
-//import java.util.List;
-//
-//import static org.hamcrest.MatcherAssert.assertThat;
-//import static org.hamcrest.core.IsEqual.equalTo;
-//import static org.hamcrest.core.Is.is;
-//import static org.hamcrest.core.IsNull.nullValue;
-//import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.deriveEffectiveConfigs;
-//import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.initialPollWaitingForConsumerGroupJoin;
-//import static org.mockito.Matchers.any;
-//
-//public class KafkaConsumerHelperTest {
-//
-//    ConsumerCommonConfigs consumerCommon;
-//    ConsumerLocalConfigs consumerLocal;
-//
-//    @Rule
-//    public ExpectedException expectedException = ExpectedException.none();
-//
-//    @Test
-//    public void test_syncAsyncTrueCommon() throws Exception {
-//        consumerCommon = new ConsumerCommonConfigs(true, true, "aTestFile", "JSON", true, 3, 50L, "");
-//
-//        expectedException.expectMessage("Both commitSync and commitAsync can not be true");
-//        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(null, consumerCommon);
-//    }
-//
-//    @Test
-//    public void test_syncAsyncTrueLocal() throws Exception {
-//        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L, "");
-//        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, true, false, 3, 50L, "1,0,test-topic");
-//        ConsumerLocalConfigsWrap localConfigsWrap = new ConsumerLocalConfigsWrap(consumerLocal);
-//
-//        expectedException.expectMessage("Both commitSync and commitAsync can not be true");
-//        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
-//    }
-//
-//    @Test
-//    public void test_effectiveConfigsIsLocal() throws Exception {
-//
-//        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L, "");
-//        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 150L, "1,0,test-topic");
-//
-//        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
-//
-//        assertThat(consumerEffectiveConfigs.getFileDumpTo(), is("sTestLocalFile"));
-//        assertThat(consumerEffectiveConfigs.getCommitAsync(), is(true));
-//        assertThat(consumerEffectiveConfigs.getCommitSync(), is(false));
-//        assertThat(consumerEffectiveConfigs.getShowRecordsConsumed(), is(false));
-//        assertThat(consumerEffectiveConfigs.getMaxNoOfRetryPollsOrTimeouts(), is(3));
-//        assertThat(consumerEffectiveConfigs.getPollingTime(), is(consumerLocal.getPollingTime()));
-//    }
-//
-//    @Test
-//    public void test_effectiveConfigsIsCentral() throws Exception {
-//
-//        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L, "");
-//        consumerLocal = null;
-//
-//        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
-//
-//        assertThat(consumerEffectiveConfigs.getFileDumpTo(), is("aTestFile"));
-//        assertThat(consumerEffectiveConfigs.getCommitAsync(), is(false));
-//        assertThat(consumerEffectiveConfigs.getCommitSync(), is(true));
-//        assertThat(consumerEffectiveConfigs.getShowRecordsConsumed(), is(true));
-//        assertThat(consumerEffectiveConfigs.getPollingTime(), is(consumerCommon.getPollingTime()));
-//    }
-//
-//    @Test
-//    public void test_effectiveCommitAsync_true() throws Exception {
-//
-//        consumerCommon = new ConsumerCommonConfigs(true, null, "aTestFile", "JSON", true, 3, 50L, "");
-//        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 50L, "1,0,test-topic");
-//
-//        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
-//
-//        assertThat(consumerEffectiveConfigs.getCommitAsync(), is(true));
-//        assertThat(consumerEffectiveConfigs.getCommitSync(), is(false));
-//    }
-//
-//    @Test
-//    public void test_effectiveCommitSync_true() throws Exception {
-//
-//        consumerCommon = new ConsumerCommonConfigs(null, true, "aTestFile", "JSON", true, 3, 50L, "");
-//        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", null, true, false, 3, 50L, "1,0,test-topic");
-//
-//        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
-//
-//        assertThat(consumerEffectiveConfigs.getCommitSync(), is(true));
-//        assertThat(consumerEffectiveConfigs.getCommitAsync(), nullValue());
-//    }
-//
-//    @Test
-//    public void test_effectiveCommitSyncFromCommon_true() throws Exception {
-//
-//        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L, "");
-//        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", null, null, false, 3, 50L, "1,0,test-topic");
-//
-//        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
-//
-//        assertThat(consumerEffectiveConfigs.getCommitSync(), is(true));
-//        assertThat(consumerEffectiveConfigs.getCommitAsync(), is(false));
-//    }
-//
-//    @Test
-//    public void test_effectiveCommitAsyncFromCommon_true() throws Exception {
-//
-//        consumerCommon = new ConsumerCommonConfigs(null, true, "aTestFile", "JSON", true, 3, 50L, "");
-//        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 150L, "1,0,test-topic");
-//
-//        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
-//
-//        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),null);
-//
-//        // then
-//        Assert.assertEquals(1, consumerJsonRecords.size());
-//        ConsumerJsonRecord consumerJsonRecord = consumerJsonRecords.get(0);
-//        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());
-//    }
-//
-//    @Test
-//    public void test_firstPoll_exits_early_on_assignment() {
-//
-//        // given
-//        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 1000L, "");
-//        consumerLocal = null;
-//        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
-//        Consumer consumer = Mockito.mock(Consumer.class);
-//        HashSet<TopicPartition> partitions = new HashSet<>();
-//        partitions.add(new TopicPartition("test.topic", 0));
-//        Mockito.when(consumer.assignment()).thenReturn(partitions);
-//
-//        // when
-//        ConsumerRecords records = initialPollWaitingForConsumerGroupJoin(consumer, consumerEffectiveConfigs);
-//
-//        // then
-//        assertThat(records.isEmpty(), is(true));
-//    }
-//
-//    @Test
-//    public void test_firstPoll_exits_on_receiving_records() {
-//
-//        // given
-//        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 5000L, "");
-//        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 50L, "1,0,test-topic");
-//        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
-//        Consumer consumer = Mockito.mock(Consumer.class);
-//        Mockito.when(consumer.assignment()).thenReturn(new HashSet<TopicPartition>());
-//
-//        ConsumerRecords consumerRecords = Mockito.mock(ConsumerRecords.class);
-//        Mockito.when(consumer.poll(any(Duration.class))).thenReturn(consumerRecords);
-//
-//        Mockito.when(consumerRecords.isEmpty()).thenReturn(false);
-//
-//        // when
-//        ConsumerRecords records = initialPollWaitingForConsumerGroupJoin(consumer, consumerEffectiveConfigs);
-//
-//        // then
-//        assertThat(records, equalTo(consumerRecords));
-//    }
-//
-//
-//    @Test
-//    public void test_firstPoll_throws_after_timeout() throws Exception {
-//
-//        // given
-//        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, null, "");
-//        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 50L, "1,0,test-topic");
-//        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
-//
-//        Consumer consumer = Mockito.mock(Consumer.class);
-//        Mockito.when(consumer.assignment()).thenReturn(new HashSet<TopicPartition>());
-//
-//        ConsumerRecords consumerRecords = Mockito.mock(ConsumerRecords.class);
-//        Mockito.when(consumer.poll(any(Duration.class))).thenReturn(consumerRecords);
-//
-//        Mockito.when(consumerRecords.isEmpty()).thenReturn(true);
-//
-//        // should throw
-//        expectedException.expectMessage("Kafka Consumer unable to join in time");
-//
-//        // when
-//        ConsumerRecords records = initialPollWaitingForConsumerGroupJoin(consumer, consumerEffectiveConfigs);
-//    }
-//}
+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;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+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.Mockito;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.Is.is;
+import static org.hamcrest.core.IsEqual.equalTo;
+import static org.hamcrest.core.IsNull.nullValue;
+import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.deriveEffectiveConfigs;
+import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.initialPollWaitingForConsumerGroupJoin;
+import static org.mockito.Matchers.any;
+
+public class KafkaConsumerHelperTest {
+
+    ConsumerCommonConfigs consumerCommon;
+    ConsumerLocalConfigs consumerLocal;
+
+    @Rule
+    public ExpectedException expectedException = ExpectedException.none();
+
+    @Test
+    public void test_syncAsyncTrueCommon() throws Exception {
+        consumerCommon = new ConsumerCommonConfigs(true, true, "aTestFile",
+                "JSON", true, 3, 50L,
+                false,"$JSON.Path", "");
+
+        expectedException.expectMessage("Both commitSync and commitAsync can not be true");
+        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(null, consumerCommon);
+    }
+
+    @Test
+    public void test_syncAsyncTrueLocal() throws Exception {
+        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path", "");
+        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, true, false, 3, 50L,false,"$JSON.Path", "1,0,test-topic");
+        ConsumerLocalConfigsWrap localConfigsWrap = new ConsumerLocalConfigsWrap(consumerLocal);
+
+        expectedException.expectMessage("Both commitSync and commitAsync can not be true");
+        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
+    }
+
+    @Test
+    public void test_effectiveConfigsIsLocal() throws Exception {
+
+        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path", "");
+        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 150L,false,"$JSON.Path", "1,0,test-topic");
+
+        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
+
+        assertThat(consumerEffectiveConfigs.getFileDumpTo(), is("sTestLocalFile"));
+        assertThat(consumerEffectiveConfigs.getCommitAsync(), is(true));
+        assertThat(consumerEffectiveConfigs.getCommitSync(), is(false));
+        assertThat(consumerEffectiveConfigs.getShowRecordsConsumed(), is(false));
+        assertThat(consumerEffectiveConfigs.getMaxNoOfRetryPollsOrTimeouts(), is(3));
+        assertThat(consumerEffectiveConfigs.getPollingTime(), is(consumerLocal.getPollingTime()));
+    }
+
+    @Test
+    public void test_effectiveConfigsIsCentral() throws Exception {
+
+        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path", "");
+        consumerLocal = null;
+
+        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
+
+        assertThat(consumerEffectiveConfigs.getFileDumpTo(), is("aTestFile"));
+        assertThat(consumerEffectiveConfigs.getCommitAsync(), is(false));
+        assertThat(consumerEffectiveConfigs.getCommitSync(), is(true));
+        assertThat(consumerEffectiveConfigs.getShowRecordsConsumed(), is(true));
+        assertThat(consumerEffectiveConfigs.getPollingTime(), is(consumerCommon.getPollingTime()));
+    }
+
+    @Test
+    public void test_effectiveCommitAsync_true() throws Exception {
+
+        consumerCommon = new ConsumerCommonConfigs(true, null, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path", "");
+        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 50L,false,"$JSON.Path", "1,0,test-topic");
+
+        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
+
+        assertThat(consumerEffectiveConfigs.getCommitAsync(), is(true));
+        assertThat(consumerEffectiveConfigs.getCommitSync(), is(false));
+    }
+
+    @Test
+    public void test_effectiveCommitSync_true() throws Exception {
+
+        consumerCommon = new ConsumerCommonConfigs(null, true, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path", "");
+        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", null, true, false, 3, 50L,false,"$JSON.Path", "1,0,test-topic");
+
+        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
+
+        assertThat(consumerEffectiveConfigs.getCommitSync(), is(true));
+        assertThat(consumerEffectiveConfigs.getCommitAsync(), nullValue());
+    }
+
+    @Test
+    public void test_effectiveCommitSyncFromCommon_true() throws Exception {
+
+        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path", "");
+        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", null, null, false, 3, 50L,false,"$JSON.Path", "1,0,test-topic");
+
+        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
+
+        assertThat(consumerEffectiveConfigs.getCommitSync(), is(true));
+        assertThat(consumerEffectiveConfigs.getCommitAsync(), is(false));
+    }
+
+    @Test
+    public void test_effectiveCommitAsyncFromCommon_true() throws Exception {
+
+        consumerCommon = new ConsumerCommonConfigs(null, true, "aTestFile", "JSON", true, 3, 50L,false,"$JSON.Path", "");
+        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 150L,false,"$JSON.Path", "1,0,test-topic");
+
+        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
+
+        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),null);
+
+        // then
+        Assert.assertEquals(1, consumerJsonRecords.size());
+        ConsumerJsonRecord consumerJsonRecord = consumerJsonRecords.get(0);
+        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());
+    }
+
+    @Test
+    public void test_firstPoll_exits_early_on_assignment() {
+
+        // given
+        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 1000L,false,"$JSON.Path", "");
+        consumerLocal = null;
+        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
+        Consumer consumer = Mockito.mock(Consumer.class);
+        HashSet<TopicPartition> partitions = new HashSet<>();
+        partitions.add(new TopicPartition("test.topic", 0));
+        Mockito.when(consumer.assignment()).thenReturn(partitions);
+
+        // when
+        ConsumerRecords records = initialPollWaitingForConsumerGroupJoin(consumer, consumerEffectiveConfigs);
+
+        // then
+        assertThat(records.isEmpty(), is(true));
+    }
+
+    @Test
+    public void test_firstPoll_exits_on_receiving_records() {
+
+        // given
+        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 5000L,false,"$JSON.Path", "");
+        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 50L,false,"$JSON.Path", "1,0,test-topic");
+        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
+        Consumer consumer = Mockito.mock(Consumer.class);
+        Mockito.when(consumer.assignment()).thenReturn(new HashSet<TopicPartition>());
+
+        ConsumerRecords consumerRecords = Mockito.mock(ConsumerRecords.class);
+        Mockito.when(consumer.poll(any(Duration.class))).thenReturn(consumerRecords);
+
+        Mockito.when(consumerRecords.isEmpty()).thenReturn(false);
+
+        // when
+        ConsumerRecords records = initialPollWaitingForConsumerGroupJoin(consumer, consumerEffectiveConfigs);
+
+        // then
+        assertThat(records, equalTo(consumerRecords));
+    }
+
+
+    @Test
+    public void test_firstPoll_throws_after_timeout() throws Exception {
+
+        // given
+        consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, null,false,"$JSON.Path", "");
+        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 50L,false,"$JSON.Path", "1,0,test-topic");
+        ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
+
+        Consumer consumer = Mockito.mock(Consumer.class);
+        Mockito.when(consumer.assignment()).thenReturn(new HashSet<TopicPartition>());
+
+        ConsumerRecords consumerRecords = Mockito.mock(ConsumerRecords.class);
+        Mockito.when(consumer.poll(any(Duration.class))).thenReturn(consumerRecords);
+
+        Mockito.when(consumerRecords.isEmpty()).thenReturn(true);
+
+        // should throw
+        expectedException.expectMessage("Kafka Consumer unable to join in time");
+
+        // when
+        ConsumerRecords records = initialPollWaitingForConsumerGroupJoin(consumer, consumerEffectiveConfigs);
+    }
+}
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 dad44351..8fb7c330 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
@@ -9,6 +9,8 @@ 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.latest.KafkaConsumeLatestExistingTopicTest;
+import org.jsmart.zerocode.integration.tests.kafka.consume.latest.KafkaConsumeLatestTest;
 import org.jsmart.zerocode.integration.tests.kafka.consume.negative.KafkaConsumeAvroNegativeTest;
 import org.jsmart.zerocode.integration.tests.kafka.produce.KafkaProduceAsyncTest;
 import org.jsmart.zerocode.integration.tests.kafka.produce.KafkaProduceIntKeyTest;
@@ -57,7 +59,9 @@ import org.junit.runners.Suite;
         KafkaProduceSyncWrongFileNameTest.class,
         KafkaConsumeSeekOffsetTest.class,
         KafkaKsqlTest.class,
-        KafkaProtobufTest.class
+        KafkaProtobufTest.class,
+        KafkaConsumeLatestTest.class,
+        KafkaConsumeLatestExistingTopicTest.class
 })
 @RunWith(Suite.class)
 public class KafkaSuite {
diff --git a/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/latest/KafkaConsumeLatestExistingTopicTest.java b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/latest/KafkaConsumeLatestExistingTopicTest.java
new file mode 100644
index 00000000..f0897de0
--- /dev/null
+++ b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/latest/KafkaConsumeLatestExistingTopicTest.java
@@ -0,0 +1,22 @@
+package org.jsmart.zerocode.integration.tests.kafka.consume.latest;
+
+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_latest.properties")
+@RunWith(ZeroCodeUnitRunner.class)
+public class KafkaConsumeLatestExistingTopicTest {
+
+    @Test
+    @Scenario("kafka/consume/latest/test_offset_to_latest_all_partitions_existing_topic.json")
+    public void testKafkaConsume_resetToLatestOffsetExistingTopic() throws Exception {
+    }
+
+    @Test
+    @Scenario("kafka/consume/latest/test_kafka_produce_consume_only_new_msg_existing_topic.json")
+    public void testKafkaProduceConsume() throws Exception {
+    }
+}
diff --git a/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/latest/KafkaConsumeLatestTest.java b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/latest/KafkaConsumeLatestTest.java
new file mode 100644
index 00000000..2d51d2a1
--- /dev/null
+++ b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/latest/KafkaConsumeLatestTest.java
@@ -0,0 +1,22 @@
+package org.jsmart.zerocode.integration.tests.kafka.consume.latest;
+
+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_latest.properties")
+@RunWith(ZeroCodeUnitRunner.class)
+public class KafkaConsumeLatestTest {
+
+    @Test
+    @Scenario("kafka/consume/latest/test_offset_to_latest_all_partitions.json")
+    public void testKafkaConsume_resetToLatestOffset() throws Exception {
+    }
+
+    @Test
+    @Scenario("kafka/consume/latest/test_kafka_produce_consume_only_new_msg.json")
+    public void testKafkaProduceConsume() throws Exception {
+    }
+}
diff --git a/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/produce/KafkaProduceToPartitionTest.java b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/produce/KafkaProduceToPartitionTest.java
index 53232507..543596be 100644
--- a/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/produce/KafkaProduceToPartitionTest.java
+++ b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/produce/KafkaProduceToPartitionTest.java
@@ -6,6 +6,7 @@ import org.jsmart.zerocode.core.runner.ZeroCodeUnitRunner;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
+
 @TargetEnv("kafka_servers/kafka_test_server.properties")
 @RunWith(ZeroCodeUnitRunner.class)
 public class KafkaProduceToPartitionTest {
diff --git a/kafka-testing/src/test/resources/kafka/consume/latest/test_kafka_produce_consume_only_new_msg.json b/kafka-testing/src/test/resources/kafka/consume/latest/test_kafka_produce_consume_only_new_msg.json
new file mode 100755
index 00000000..65c59228
--- /dev/null
+++ b/kafka-testing/src/test/resources/kafka/consume/latest/test_kafka_produce_consume_only_new_msg.json
@@ -0,0 +1,35 @@
+{
+    "scenarioName": "Simple produce and consume - only the new message",
+    "steps": [
+        {
+            "name": "send_to_kafka",
+            "url": "kafka-topic:local-demo-topic",
+            "operation": "PRODUCE",
+            "request": {
+                "records":[
+                    {
+                        "key": "${RANDOM.NUMBER}",
+                        "value": "Hello - A New Message 101"
+                    }
+                ]
+            },
+            "assertions": {
+                "status" : "Ok"
+            }
+        },
+        {
+            "name": "get_from_kafka",
+            "url": "kafka-topic:local-demo-topic",
+            "operation": "CONSUME",
+            "request": {
+                "consumerLocalConfigs": {
+                    "maxNoOfRetryPollsOrTimeouts": 3,
+                    "commitSync": true
+                }
+            },
+            "assertions": {
+                "size" : "$GT.0"
+            }
+        }
+    ]
+}
diff --git a/kafka-testing/src/test/resources/kafka/consume/latest/test_kafka_produce_consume_only_new_msg_existing_topic.json b/kafka-testing/src/test/resources/kafka/consume/latest/test_kafka_produce_consume_only_new_msg_existing_topic.json
new file mode 100755
index 00000000..82748520
--- /dev/null
+++ b/kafka-testing/src/test/resources/kafka/consume/latest/test_kafka_produce_consume_only_new_msg_existing_topic.json
@@ -0,0 +1,37 @@
+{
+    "scenarioName": "Simple produce and consume - only the new message",
+    "steps": [
+        {
+            "name": "send_to_kafka",
+            "url": "kafka-topic:demo-c1",
+//            "url": "kafka-topic:local-demo-topic", //<--- This will work becaz of same topic as the previous test scenario
+            "operation": "PRODUCE",
+            "request": {
+                "records":[
+                    {
+                        "key": "${RANDOM.NUMBER}",
+                        "value": "Hello - A New Message 101"
+                    }
+                ]
+            },
+            "assertions": {
+                "status" : "Ok"
+            }
+        },
+        {
+            "name": "get_from_kafka",
+            "url": "kafka-topic:demo-c1",
+//            "url": "kafka-topic:local-demo-topic", //<--- This will work becaz of same topic as the previous test scenario
+            "operation": "CONSUME",
+            "request": {
+                "consumerLocalConfigs": {
+                    "maxNoOfRetryPollsOrTimeouts": 3,
+                    "commitSync": true
+                }
+            },
+            "assertions": {
+                "size" : "$GT.0"
+            }
+        }
+    ]
+}
diff --git a/kafka-testing/src/test/resources/kafka/consume/latest/test_offset_to_latest_all_partitions.json b/kafka-testing/src/test/resources/kafka/consume/latest/test_offset_to_latest_all_partitions.json
new file mode 100755
index 00000000..2620b1d0
--- /dev/null
+++ b/kafka-testing/src/test/resources/kafka/consume/latest/test_offset_to_latest_all_partitions.json
@@ -0,0 +1,35 @@
+{
+    "scenarioName": "Reset offset to latest 1st",
+    "steps": [
+        {
+            "name": "just_to_auto_create_the_topic", //<--- Otherwise this step is not needed
+            "url": "kafka-topic:local-demo-topic",
+            "operation": "PRODUCE",
+            "request": {
+                "records": [
+                    {
+                        "key": "${RANDOM.NUMBER}",
+                        "value": "Hello - I am a Message. I need a topic please"
+                    }
+                ]
+            },
+            "assertions": {
+                "status": "Ok"
+            }
+        },
+        {
+            "name": "reset_now",
+            "url": "kafka-topic:local-demo-topic",
+            "operation": "CONSUME",
+            "request": {
+                "consumerLocalConfigs": {
+                    "maxNoOfRetryPollsOrTimeouts": 3,
+                    "commitSync": true
+                }
+            },
+            "assertions": {
+                "size": 0
+            }
+        }
+    ]
+}
diff --git a/kafka-testing/src/test/resources/kafka/consume/latest/test_offset_to_latest_all_partitions_existing_topic.json b/kafka-testing/src/test/resources/kafka/consume/latest/test_offset_to_latest_all_partitions_existing_topic.json
new file mode 100755
index 00000000..6ed7f7fb
--- /dev/null
+++ b/kafka-testing/src/test/resources/kafka/consume/latest/test_offset_to_latest_all_partitions_existing_topic.json
@@ -0,0 +1,20 @@
+{
+    "scenarioName": "Reset offset to latest 1st",
+    "steps": [
+        {
+            "name": "reset_now",
+            "url": "kafka-topic:demo-c1",
+//            "url": "kafka-topic:local-demo-topic", //<--- This will work becaz of same topic as the previous test scenario
+            "operation": "CONSUME",
+            "request": {
+                "consumerLocalConfigs": {
+                    "maxNoOfRetryPollsOrTimeouts": 3,
+                    "commitSync": true
+                }
+            },
+            "assertions": {
+                "size": 0
+            }
+        }
+    ]
+}
diff --git a/kafka-testing/src/test/resources/kafka/produce/test_kafka_produce_to_partition.json b/kafka-testing/src/test/resources/kafka/produce/test_kafka_produce_to_partition.json
index edb2fadb..b565ccec 100755
--- a/kafka-testing/src/test/resources/kafka/produce/test_kafka_produce_to_partition.json
+++ b/kafka-testing/src/test/resources/kafka/produce/test_kafka_produce_to_partition.json
@@ -23,24 +23,32 @@
                     }
                 }
             }
-        },
-        {
-            "name": "load_kafka_wrong_partition",
-            "url": "kafka-topic:demo-4",
-            "operation": "produce",
-            "request": {
-                "records":[
-                    {
-                        "key": "${RANDOM.NUMBER}",
-                        "value": "Hello World",
-                        "partition": 9
-                    }
-                ]
-            },
-            "assertions": {
-                "status" : "Failed",
-                "message" : "Invalid partition given with record: 9 is not in the range [0...1)."
-            }
         }
+        // This works.
+        // But takes 60 secs to get response for the timeout. See more comments below
+//        {
+//            "name": "load_kafka_wrong_partition",
+//            "url": "kafka-topic:demo-4",
+//            "operation": "produce",
+//            "request": {
+//                "records":[
+//                    {
+//                        "key": "${RANDOM.NUMBER}",
+//                        "value": "Hello World",
+//                        "partition": 9
+//                    }
+//                ]
+//            },
+//            "assertions": {
+//                // This works. But takes 60 secs to get response for the timeout
+//                // This is after upgrading to the new version of Kafka client. 3.3.1 (version.kafka-clients)
+//                "status" : "Failed",
+//                "message" : "org.apache.kafka.common.errors.TimeoutException: Topic demo-4 not present in metadata after 60000 ms."
+//
+//                // Old client version.kafka-clients=2.1.0
+//                // "status" : "Failed",
+//                // "message" : "Invalid partition given with record: 9 is not in the range [0...1)."
+//            }
+//        }
     ]
 }
diff --git a/kafka-testing/src/test/resources/kafka_servers/kafka_consumer_latest.properties b/kafka-testing/src/test/resources/kafka_servers/kafka_consumer_latest.properties
new file mode 100755
index 00000000..6f66b596
--- /dev/null
+++ b/kafka-testing/src/test/resources/kafka_servers/kafka_consumer_latest.properties
@@ -0,0 +1,37 @@
+# =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
+#          kafka consumer properties
+# =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
+#group.id=consumerGroup_10
+group.id=consumerGroup_${GLOBAL.RANDOM.NUMBER}
+key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
+value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
+max.poll.records=2
+enable.auto.commit=false
+auto.offset.reset=latest
+
+# Both has to be present or max.poll.inteval.ms can be present alone.
+# This doesn't help in rebalance anyway, so don't fully rely on this property
+#session.timeout.ms=10000
+max.poll.inteval.ms=5000
+
+# -----------------------------
+# client.id is auto generated. Making it unique will have no effect if they belong to same group.
+# Making the group.id as unique makes sense and the new group ca consume same records once again.
+# client.id uniqueness will differentiate from another consumer in the same group.
+# Refer : ConsumerConfig.java in the source code.
+# /kafka/kafka/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
+# -----------------------------
+client.id=consumer-${RANDOM.NUMBER}
+#group.id=None
+#enable.auto.commit=true
+#key.deserializer=org.apache.kafka.common.serialization.LongDeserializer
+#value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
+#
+## fast session timeout makes it more fun to play with failover
+#
+## These buffer sizes seem to be needed to avoid consumer switching to
+## a mode where it processes one bufferful every 5 seconds with multiple
+## timeouts along the way.  No idea why this happens.
+#fetch.min.bytes=50000
+#receive.buffer.bytes=262144
+#max.partition.fetch.bytes=2097152
\ No newline at end of file
diff --git a/kafka-testing/src/test/resources/kafka_servers/kafka_test_server_latest.properties b/kafka-testing/src/test/resources/kafka_servers/kafka_test_server_latest.properties
new file mode 100755
index 00000000..15c08f39
--- /dev/null
+++ b/kafka-testing/src/test/resources/kafka_servers/kafka_test_server_latest.properties
@@ -0,0 +1,43 @@
+# =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
+# kafka bootstrap servers comma separated
+# e.g. localhost:9092,host2:9093
+# =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
+kafka.bootstrap.servers=localhost:9092
+
+kafka.producer.properties=kafka_servers/kafka_producer_unique.properties
+#consumer with latest config (note-producer config doesn't need any changes)
+kafka.consumer.properties=kafka_servers/kafka_consumer_latest.properties
+
+# --------------------------------------------------------------------
+# Optional local consumer properties common/central to all test cases.
+# These can be overwritten by the tests locally.
+# --------------------------------------------------------------------
+# If this property is set, then the consumer does a commitSync after reading the message(s)
+# Make sure you don't set both commitSync and commitAsync to true
+consumer.commitSync = true
+# If this property is set, then the consumer does a commitAsync after reading the message(s)
+# Make sure you don't set both commitSync and commitAsync to true
+consumer.commitAsync = false
+# All records those were read are dumped to this specified file path
+# This path can be a relative path or an absolute path. If the file
+# does not exist, it creates the file and dumps the records
+consumer.fileDumpTo= target/temp/demo.txt
+# If this property is set to true, all records are shown in the response.
+# When dealing with large number of records, you might not be interested
+# in the individual records, but interested in the recordCount
+# i.e. total number of records consumed
+consumer.showRecordsConsumed=false
+# That means if any record(s) are read, then this counter is reset to 0(zero) and the consumer
+# polls again. So if no records are fetched for a specific poll interval, then the consumer
+# gives a retry retrying until this max number polls/reties reached.
+consumer.maxNoOfRetryPollsOrTimeouts = 5
+# Polling time in milli seconds i.e how long the consumer should poll before
+# the next retry poll
+consumer.pollingTime = 1000
+
+# Whether same consumer which was created earlier for other test-steps or scenarios to be used.
+# If set to false or if this config is not present, it creates a new Consumer for every tests
+consumer.cacheByTopic=true
+
+# local producer properties
+producer.key1=value1-testv ycvb
diff --git a/kafka-testing/src/test/resources/kafka_servers/kafka_test_server_unique.properties b/kafka-testing/src/test/resources/kafka_servers/kafka_test_server_unique.properties
index 8597d83d..9b0992f9 100755
--- a/kafka-testing/src/test/resources/kafka_servers/kafka_test_server_unique.properties
+++ b/kafka-testing/src/test/resources/kafka_servers/kafka_test_server_unique.properties
@@ -33,6 +33,5 @@ consumer.maxNoOfRetryPollsOrTimeouts = 5
 # Polling time in milli seconds i.e how long the consumer should poll before
 # the next retry poll
 consumer.pollingTime = 1000
-
 # local producer properties
 producer.key1=value1-testv ycvb