Repository: authorjapps/zerocode
Pull request #422 (2020-08-22 15:49:45)

Fix issue 421 kafka receiver
# Add initialPoll to wait for consumer group join

[Link to Branch
](https://github.com/ddingel/zerocode/tree/fix-issue-421-KafkaReceiver)

## Motivation and Context
In a kafka consume step, on the first poll, a consumer group join will be triggered. 
This can take multiple seconds to complete. Currently we use will use our maxNoOfRetryPollsOrTimeouts in that time.
If the consumer group join has not finished we return without any result or meaningful message.

The proposed change will move the initial poll out of the loop.
Additionally if the initial poll fails an exception will be thrown.

## Checklist:

* [ X] Unit tests added

* [x] Integration tests added

* [ X] Test names are meaningful

* [ X] Feature manually tested

* [x] Branch build passed

* [ X] No 'package.*' in the imports

* [ ] Relevant Wiki page updated with clear instruction for the end user
  * [ ] Not applicable. This was only a refactor change, no functional or behaviour changes were introduced

* [ ] Http test added to `http-testing` module(if applicable) ?
  * [X ] Not applicable. The changes did not affect HTTP automation flow

* [ X] Kafka test added to `kafka-testing` module(if applicable) ?
  * [ ] Not applicable. The changes did not affect Kafka automation flow



Patch:
diff --git a/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaConsumerHelper.java b/core/src/main/java/org/jsmart/zerocode/core/kafka/helper/KafkaConsumerHelper.java
index 27bb9e71..a9255b36 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
@@ -18,6 +18,7 @@ import java.util.Set;
 
 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.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.header.Header;
@@ -33,6 +34,9 @@ import org.jsmart.zerocode.core.kafka.receive.message.ConsumerRawRecords;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+
 import static java.lang.Integer.parseInt;
 import static java.lang.Long.parseLong;
 import static java.util.Optional.ofNullable;
@@ -67,6 +71,20 @@ public class KafkaConsumerHelper {
         }
     }
 
+    public static ConsumerRecords initialPollWaitingForConsumerGroupJoin(Consumer consumer) {
+        for (int run = 0; run < 10; run++) {
+            if (!consumer.assignment().isEmpty()) {
+                return new ConsumerRecords(new HashMap());
+            }
+            ConsumerRecords records = consumer.poll(Duration.of(500, ChronoUnit.MILLIS));
+            if (!records.isEmpty()) {
+                return records;
+            }
+        }
+
+        throw new RuntimeException("\n********* Kafka Consumer unable to join in time *********\n");
+    }
+
     public static void validateLocalConfigs(ConsumerLocalConfigs localConfigs) {
         if (localConfigs != null) {
             Boolean localCommitSync = localConfigs.getCommitSync();
@@ -286,9 +304,9 @@ public class KafkaConsumerHelper {
 
     private static void validateSeekConfig(ConsumerLocalConfigs localConfigs) {
         String seek = localConfigs.getSeek();
-        if(!isEmpty(seek)) {
+        if (!isEmpty(seek)) {
             String[] split = seek.split(",");
-            if(split == null || split.length < 3) {
+            if (split == null || split.length < 3) {
                 throw new RuntimeException("\n------> 'seek' should contain 'topic,partition,offset' e.g. 'topic1,0,2' ");
             }
         }
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 4df520b8..4201fab8 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
@@ -1,6 +1,5 @@
 package org.jsmart.zerocode.core.kafka.receive;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import com.google.inject.name.Named;
@@ -11,7 +10,6 @@ import java.util.List;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.jsmart.zerocode.core.di.provider.ObjectMapperProvider;
 import org.jsmart.zerocode.core.kafka.consume.ConsumerLocalConfigs;
 import org.jsmart.zerocode.core.kafka.receive.message.ConsumerJsonRecord;
 import org.slf4j.Logger;
@@ -26,6 +24,7 @@ import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.getMaxTi
 import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.getPollTime;
 import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.handleCommitSyncAsync;
 import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.handleSeekOffset;
+import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.initialPollWaitingForConsumerGroupJoin;
 import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.prepareResult;
 import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.readConsumerLocalTestProperties;
 import static org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper.readJson;
@@ -37,8 +36,6 @@ public class KafkaReceiver {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(KafkaReceiver.class);
 
-    private final ObjectMapper objectMapper = new ObjectMapperProvider().get();
-
     @Inject(optional = true)
     @Named("kafka.consumer.properties")
     private String consumerPropertyFile;
@@ -63,48 +60,34 @@ public class KafkaReceiver {
 
         handleSeekOffset(effectiveLocal, consumer);
 
-        while (true) {
-            LOGGER.info("polling records  - noOfTimeOuts reached : " + noOfTimeOuts);
-
-            final ConsumerRecords records = consumer.poll(ofMillis(getPollTime(effectiveLocal)));
-            noOfTimeOuts++;
+        LOGGER.info("initial polling to trigger ConsumerGroupJoin");
 
-            if (records.count() == 0) {
-                if (noOfTimeOuts >= getMaxTimeOuts(effectiveLocal)) {
-                    break;
-                } else {
-                    continue;
-                }
-            } else {
-                LOGGER.info("Received {} records after {} timeouts\n", records.count(), noOfTimeOuts);
-            }
+        ConsumerRecords records = initialPollWaitingForConsumerGroupJoin(consumer);
 
-            if (records != null) {
-                Iterator recordIterator = records.iterator();
+        if(!records.isEmpty()) {
+            LOGGER.info("Received {} records on initial poll\n", records.count());
 
-                LOGGER.info("Consumer chosen recordType: " + effectiveLocal.getRecordType());
+            appendNewRecords(records, rawRecords, jsonRecords, effectiveLocal);
 
-                switch (effectiveLocal.getRecordType()) {
-                    case RAW:
-                        readRaw(rawRecords, recordIterator);
-                        break;
+            handleCommitSyncAsync(consumer, consumerCommonConfigs, effectiveLocal);
+        }
 
-                    case JSON:
-                        readJson(jsonRecords, recordIterator);
-                        break;
+        while (noOfTimeOuts < getMaxTimeOuts(effectiveLocal)) {
+            LOGGER.info("polling records  - noOfTimeOuts reached : " + noOfTimeOuts);
 
-                    default:
-                        throw new RuntimeException("Unsupported record type - '" + effectiveLocal.getRecordType()
-                                + "'. Supported values are 'JSON','RAW'");
-                }
+            records = consumer.poll(ofMillis(getPollTime(effectiveLocal)));
+            noOfTimeOuts++;
 
+            if (records.count() == 0) {
+                continue;
             }
 
+            LOGGER.info("Received {} records after {} timeouts\n", records.count(), noOfTimeOuts);
+
+            appendNewRecords(records, rawRecords, jsonRecords, effectiveLocal);
+
             handleCommitSyncAsync(consumer, consumerCommonConfigs, effectiveLocal);
 
-            if (noOfTimeOuts >= getMaxTimeOuts(effectiveLocal)) {
-                break;
-            }
         }
 
         consumer.close();
@@ -115,4 +98,24 @@ public class KafkaReceiver {
 
     }
 
+    private void appendNewRecords(ConsumerRecords records, List<ConsumerRecord> rawRecords, List<ConsumerJsonRecord> jsonRecords, ConsumerLocalConfigs effectiveLocal) throws IOException {
+        Iterator recordIterator = records.iterator();
+
+        LOGGER.info("Consumer chosen recordType: " + effectiveLocal.getRecordType());
+
+        switch (effectiveLocal.getRecordType()) {
+            case RAW:
+                readRaw(rawRecords, recordIterator);
+                break;
+
+            case JSON:
+                readJson(jsonRecords, recordIterator);
+                break;
+
+            default:
+                throw new RuntimeException("Unsupported record type - '" + effectiveLocal.getRecordType()
+                        + "'. Supported values are 'JSON','RAW'");
+        }
+    }
+
 }


Test patch:
diff --git a/core/src/test/java/org/jsmart/zerocode/core/kafka/helper/KafkaConsumerHelperTest.java b/core/src/test/java/org/jsmart/zerocode/core/kafka/helper/KafkaConsumerHelperTest.java
index bae73bb3..8dd7be17 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,10 +1,11 @@
 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.hamcrest.CoreMatchers;
 import org.jsmart.zerocode.core.kafka.consume.ConsumerLocalConfigs;
 import org.jsmart.zerocode.core.kafka.consume.ConsumerLocalConfigsWrap;
 import org.jsmart.zerocode.core.kafka.receive.ConsumerCommonConfigs;
@@ -13,31 +14,33 @@ import org.junit.Assert;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
-import org.mockito.Mock;
 import org.mockito.Mockito;
 
 import java.io.IOException;
+import java.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;
 
-    //ObjectMapper objectMapper = (new ObjectMapperProvider()).get();
-
     @Rule
     public ExpectedException expectedException = ExpectedException.none();
 
     @Test
-    public void test_syncAsyncTrueCommon() throws Exception{
+    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");
@@ -45,7 +48,7 @@ public class KafkaConsumerHelperTest {
     }
 
     @Test
-    public void test_syncAsyncTrueLocal() throws Exception{
+    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);
@@ -55,7 +58,7 @@ public class KafkaConsumerHelperTest {
     }
 
     @Test
-    public void test_effectiveConfigsIsLocal() throws Exception{
+    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");
@@ -71,7 +74,7 @@ public class KafkaConsumerHelperTest {
     }
 
     @Test
-    public void test_effectiveConfigsIsCentrall() throws Exception{
+    public void test_effectiveConfigsIsCentral() throws Exception {
 
         consumerCommon = new ConsumerCommonConfigs(true, false, "aTestFile", "JSON", true, 3, 50L, "");
         consumerLocal = null;
@@ -86,10 +89,10 @@ public class KafkaConsumerHelperTest {
     }
 
     @Test
-    public void test_effectiveCommitAsync_true() throws Exception{
+    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");
+        consumerLocal = new ConsumerLocalConfigs("RAW", "sTestLocalFile", true, false, false, 3, 50L, "1,0,test-topic");
 
         ConsumerLocalConfigs consumerEffectiveConfigs = deriveEffectiveConfigs(consumerLocal, consumerCommon);
 
@@ -98,7 +101,7 @@ public class KafkaConsumerHelperTest {
     }
 
     @Test
-    public void test_effectiveCommitSync_true() throws Exception{
+    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");
@@ -110,7 +113,7 @@ public class KafkaConsumerHelperTest {
     }
 
     @Test
-    public void test_effectiveCommitSyncFromCommon_true() throws Exception{
+    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");
@@ -122,9 +125,9 @@ public class KafkaConsumerHelperTest {
     }
 
     @Test
-    public void test_effectiveCommitAsyncFromCommon_true() throws Exception{
+    public void test_effectiveCommitAsyncFromCommon_true() throws Exception {
 
-        consumerCommon = new ConsumerCommonConfigs(null, true, "aTestFile", "JSON", true, 3,50L, "");
+        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);
@@ -153,4 +156,56 @@ public class KafkaConsumerHelperTest {
         Assert.assertEquals("\"value\"", consumerJsonRecord.getValue().toString());
         Assert.assertEquals(Collections.singletonMap("headerKey", "headerValue"), consumerJsonRecord.getHeaders());
     }
+
+    @Test
+    public void test_firstPoll_exits_early_on_assignment() {
+        // given
+        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);
+
+        // then
+        assertThat(records.isEmpty(), is(true));
+    }
+
+    @Test
+    public void test_firstPoll_exits_on_receiving_records() {
+        // given
+        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);
+
+        // then
+        assertThat(records, equalTo(consumerRecords));
+    }
+
+
+    @Test
+    public void test_firstPoll_throws_after_timeout() throws Exception {
+        // given
+        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);
+    }
 }
\ No newline at end of file
diff --git a/kafka-testing/src/test/resources/kafka/consume/test_kafka_consume.json b/kafka-testing/src/test/resources/kafka/consume/test_kafka_consume.json
index be453380..2c36d5e4 100755
--- a/kafka-testing/src/test/resources/kafka/consume/test_kafka_consume.json
+++ b/kafka-testing/src/test/resources/kafka/consume/test_kafka_consume.json
@@ -22,6 +22,9 @@
             "url": "kafka-topic:demo-c1",
             "operation": "unload",
             "request": {
+                "consumerLocalConfigs": {
+                    "maxNoOfRetryPollsOrTimeouts": 1
+                }
             },
             "assertions": {
                 // this will be 1  only when consumer does a commit i.e. commitAsync or sync,