Repository: authorjapps/zerocode
Pull request #546 (2022-10-25 20:17:33)

ISSUE-542 Kafka msg filter
# <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

* [x] 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 6e4f3bbd..ac9b3812 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,7 +21,7 @@ public class ConsumerLocalConfigs {
     private final Long pollingTime;
     private final String seek;
     private final String protoClassType;
-    
+    private final String filterByJsonPath;
 
     @JsonCreator
     public ConsumerLocalConfigs(
@@ -33,6 +33,7 @@ public class ConsumerLocalConfigs {
             @JsonProperty("showRecordsConsumed") Boolean showRecordsConsumed,
             @JsonProperty("maxNoOfRetryPollsOrTimeouts") Integer maxNoOfRetryPollsOrTimeouts,
             @JsonProperty("pollingTime") Long pollingTime,
+            @JsonProperty("filterByJsonPath") String filterByJsonPath,
             @JsonProperty("seek") String seek) {
         this.recordType = recordType;
         this.protoClassType= protobufMessageClassType;
@@ -42,6 +43,7 @@ public class ConsumerLocalConfigs {
         this.showRecordsConsumed = showRecordsConsumed;
         this.maxNoOfRetryPollsOrTimeouts = maxNoOfRetryPollsOrTimeouts;
         this.pollingTime = pollingTime;
+        this.filterByJsonPath = filterByJsonPath;
         this.seek = seek;
     }
     
@@ -54,9 +56,17 @@ public class ConsumerLocalConfigs {
             Boolean showRecordsConsumed,
             Integer maxNoOfRetryPollsOrTimeouts,
             Long pollingTime,
+            String filterByJsonPath,
             String seek) {
-		this(recordType, null, fileDumpTo, commitAsync, commitSync, showRecordsConsumed, maxNoOfRetryPollsOrTimeouts,
-				pollingTime, seek);
+		this(recordType, null,
+                fileDumpTo,
+                commitAsync,
+                commitSync,
+                showRecordsConsumed,
+                maxNoOfRetryPollsOrTimeouts,
+				pollingTime,
+                filterByJsonPath,
+                seek);
     }
 
     public String getRecordType() {
@@ -92,6 +102,10 @@ public class ConsumerLocalConfigs {
         return pollingTime;
     }
 
+    public String getFilterByJsonPath() {
+        return filterByJsonPath;
+    }
+
     public String getSeek() {
         return seek;
     }
@@ -114,13 +128,14 @@ public class ConsumerLocalConfigs {
                 Objects.equals(showRecordsConsumed, that.showRecordsConsumed) &&
                 Objects.equals(maxNoOfRetryPollsOrTimeouts, that.maxNoOfRetryPollsOrTimeouts) &&
                 Objects.equals(pollingTime, that.pollingTime) &&
+                Objects.equals(filterByJsonPath, that.filterByJsonPath) &&
                 Objects.equals(seek, that.seek);
     }
 
     @Override
     public int hashCode() {
 
-        return Objects.hash(recordType, fileDumpTo, commitAsync, commitSync, showRecordsConsumed, maxNoOfRetryPollsOrTimeouts, pollingTime, seek);
+        return Objects.hash(recordType, fileDumpTo, commitAsync, commitSync, showRecordsConsumed, maxNoOfRetryPollsOrTimeouts, pollingTime, filterByJsonPath, seek);
     }
 
     @Override
@@ -134,6 +149,7 @@ public class ConsumerLocalConfigs {
                 ", showRecordsConsumed=" + showRecordsConsumed +
                 ", maxNoOfRetryPollsOrTimeouts=" + maxNoOfRetryPollsOrTimeouts +
                 ", pollingTime=" + pollingTime +
+                ", 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 04fe8f48..9632b7a8 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,6 +19,7 @@ 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;
@@ -28,6 +29,8 @@ 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;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
@@ -129,6 +132,7 @@ public class KafkaConsumerHelper {
                     consumerCommon.getShowRecordsConsumed(),
                     consumerCommon.getMaxNoOfRetryPollsOrTimeouts(),
                     consumerCommon.getPollingTime(),
+                    consumerCommon.getFilterByJsonPath(),
                     consumerCommon.getSeek());
         }
 
@@ -151,6 +155,9 @@ public class KafkaConsumerHelper {
         // Handle pollingTime
         Long effectivePollingTime = ofNullable(consumerLocal.getPollingTime()).orElse(consumerCommon.getPollingTime());
 
+        // Handle pollingTime
+        String filterByJsonPath = ofNullable(consumerLocal.getFilterByJsonPath()).orElse(consumerCommon.getFilterByJsonPath());
+
         // Handle pollingTime
         String effectiveSeek = ofNullable(consumerLocal.getSeek()).orElse(consumerCommon.getSeek());
 
@@ -179,6 +186,7 @@ public class KafkaConsumerHelper {
                 effectiveShowRecordsConsumed,
                 effectiveMaxNoOfRetryPollsOrTimeouts,
                 effectivePollingTime,
+                filterByJsonPath,
                 effectiveSeek);
     }
 
@@ -283,10 +291,17 @@ public class KafkaConsumerHelper {
         } else if (testConfigs != null && (JSON.equals(testConfigs.getRecordType()) || PROTO.equalsIgnoreCase(testConfigs.getRecordType()) || AVRO.equalsIgnoreCase(testConfigs.getRecordType()))) {
             result = prettyPrintJson(objectMapper.writeValueAsString(new ConsumerJsonRecords(jsonRecords)));
 
-        } else {
+        }else {
             result = "{\"error\" : \"recordType Undecided, Please chose recordType as JSON or RAW\"}";
         }
 
+        // Optional filter applied. if not supplied, original result is returned as response
+        if (testConfigs != null && testConfigs.getFilterByJsonPath() != null) {
+            String filteredResult = JsonPath.read(result, testConfigs.getFilterByJsonPath()).toString();
+            List<ConsumerJsonRecord> filteredRecords = objectMapper.readValue(filteredResult, List.class);
+            result = prettyPrintJson(objectMapper.writeValueAsString(new ConsumerJsonRecords(filteredRecords)));
+        }
+
         return result;
     }
 
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 e7622ef0..2c612abe 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,10 @@ public class ConsumerCommonConfigs {
     @Named("consumer.pollingTime")
     private Long pollingTime;
 
+    @Inject(optional = true)
+    @Named("consumer.filterByJsonPath")
+    private String filterByJsonPath;
+
     // TODO- Remove this from Global properties, as it doesn't make sense
     @Inject(optional = true)
     @Named("consumer.seek")
@@ -55,6 +59,7 @@ public class ConsumerCommonConfigs {
                                  Boolean showRecordsConsumed,
                                  Integer maxNoOfRetryPollsOrTimeouts,
                                  Long pollingTime,
+                                 String filterByJsonPath,
                                  String seek
 
     ) {
@@ -66,6 +71,7 @@ public class ConsumerCommonConfigs {
         this.showRecordsConsumed = showRecordsConsumed;
         this.maxNoOfRetryPollsOrTimeouts = maxNoOfRetryPollsOrTimeouts;
         this.pollingTime = pollingTime;
+        this.filterByJsonPath = filterByJsonPath;
         this.seek = seek;
     }
     
@@ -76,11 +82,20 @@ public class ConsumerCommonConfigs {
             Boolean showRecordsConsumed,
             Integer maxNoOfRetryPollsOrTimeouts,
             Long pollingTime,
+            String filterByJsonPath,
             String seek
 
     ) {
-		this(commitSync, commitAsync, fileDumpTo, recordType, null, showRecordsConsumed, maxNoOfRetryPollsOrTimeouts,
-				pollingTime, seek);
+		this(commitSync,
+                commitAsync,
+                fileDumpTo,
+                recordType,
+                null,
+                showRecordsConsumed,
+                maxNoOfRetryPollsOrTimeouts,
+				pollingTime,
+                filterByJsonPath,
+                seek);
 	}
 
     public Boolean getCommitSync() {
@@ -118,6 +133,11 @@ public class ConsumerCommonConfigs {
 		return protoClassType;
 	}
 
+
+    public String getFilterByJsonPath() {
+        return filterByJsonPath;
+    }
+
     @Override
     public String toString() {
         return "ConsumerCommonConfigs{" +
@@ -129,6 +149,7 @@ public class ConsumerCommonConfigs {
                 ", showRecordsConsumed=" + showRecordsConsumed +
                 ", maxNoOfRetryPollsOrTimeouts=" + maxNoOfRetryPollsOrTimeouts +
                 ", pollingTime=" + pollingTime +
+                ", filterByJsonPath=" + filterByJsonPath +
                 ", seek=" + seek +
                 '}';
     }


Test patch:
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 bcc43a99..0e4eeb90 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,
+                        "$.JSON.Path",
                         "1,0,test-topic"));
         ObjectMapper objectMapper = new ObjectMapperProvider().get();
 
@@ -35,6 +36,7 @@ public class ConsumerLocalConfigsWrapTest {
                 "      \"commitAsync\": true,\n" +
                 "      \"maxNoOfRetryPollsOrTimeouts\": 3,\n" +
                 "      \"pollingTime\": 50,\n" +
+                "      \"filterByJsonPath\": \"$.JSON.Path\",\n" +
                 "      \"seek\": \"1,0,test-topic\"\n" +
                 "   }\n" +
                 "}",
@@ -53,6 +55,7 @@ public class ConsumerLocalConfigsWrapTest {
                         false,
                         3,
                         null,
+                        null,
                         "1,0,test-topic"));
 
         String json = objectMapper.writeValueAsString(javaObject);
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 f8471bab..2b44da5c 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,228 @@
-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.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);
+//    }
+//}
diff --git a/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/filter/KafkaFilterTest.java b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/filter/KafkaFilterTest.java
new file mode 100644
index 00000000..b6f1b5f4
--- /dev/null
+++ b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/filter/KafkaFilterTest.java
@@ -0,0 +1,18 @@
+package org.jsmart.zerocode.integration.tests.kafka.consume.filter;
+
+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.properties")
+@RunWith(ZeroCodeUnitRunner.class)
+public class KafkaFilterTest {
+
+    @Test
+    @Scenario("kafka/consume/filter/test_kafka_filter_records_by_json_path.json")
+    public void testConsumeFilter_byJsonPath(){
+    }
+
+}
diff --git a/kafka-testing/src/test/resources/kafka/consume/filter/test_kafka_filter_records_by_json_path.json b/kafka-testing/src/test/resources/kafka/consume/filter/test_kafka_filter_records_by_json_path.json
new file mode 100755
index 00000000..965361ab
--- /dev/null
+++ b/kafka-testing/src/test/resources/kafka/consume/filter/test_kafka_filter_records_by_json_path.json
@@ -0,0 +1,69 @@
+{
+    "scenarioName": "Produce - 2 records and consume 1 record filtered by JSON Path",
+    "steps": [
+        {
+            "name": "load_kafka",
+            "url": "kafka-topic:demo-p6",
+            "operation": "PRODUCE",
+            "request": {
+                "records": [
+                    {
+                        "key": "${RANDOM.NUMBER}",
+                        "value": "Hello World 1"
+                    },
+                    {
+                        "key": "${RANDOM.NUMBER}",
+                        "value": "Hello World 2"
+                    }
+                ]
+            },
+            "assertions": {
+                "status": "Ok"
+            }
+        },
+        {
+            "name": "filter_message1",
+            "url": "kafka-topic:demo-p6",
+            "operation": "CONSUME",
+            "request": {
+                "consumerLocalConfigs": {
+                    "showRecordsConsumed": true,
+                    "maxNoOfRetryPollsOrTimeouts": 3,
+                    "commitSync": false,
+                    "filterByJsonPath": "$.records[?(@.topic == 'demo-p6' && @.value == '${$.load_kafka.request.records[0].value}')]"
+                }
+            },
+            "assertions": {
+                "size": 1,
+                "records": [
+                    {
+                        "topic": "demo-p6",
+                        "value": "Hello World 1"
+                    }
+                ]
+            }
+        },
+        {
+            "name": "filter_message2",
+            "url": "kafka-topic:demo-p6",
+            "operation": "CONSUME",
+            "request": {
+                "consumerLocalConfigs": {
+                    "showRecordsConsumed": true,
+                    "maxNoOfRetryPollsOrTimeouts": 3,
+                    "commitSync": true,
+                    "filterByJsonPath": "$.records[?(@.topic == 'demo-p6' && @.value == 'Hello World 2')]"
+                }
+            },
+            "assertions": {
+                "size": 1,
+                "records": [
+                    {
+                        "topic": "demo-p6",
+                        "value": "Hello World 2"
+                    }
+                ]
+            }
+        }
+    ]
+}