Repository: authorjapps/zerocode
Pull request #463 (2020-11-25 04:12:30)

ISSUE-462 # fix poll time out to intial consumer group join
# initialPollWaitingForConsumerGroupJoin time should be configurable

PR Branch
https://github.com/dandalavinod/zerocode/tree/issue462-polltimeout-kafkajoin

## Motivation and Context
Intial poll of the kafka consumer group is currently waiting for 500ms which in some cases, where there are network latency or broker throughput issues, fail with kafka consumer fail to join in time. 
The fix is to make the intial poll duration wait configurable to use the consumer.pollingTime field set in the properties file and if this property is not set then will default to current 500ms poll duration

## Checklist:

* [ x] Unit tests added
   updated existing unit tests to fit and behave per the changes added.

* [x] Integration tests added

* [x ] Test names are meaningful

* [ x] Feature manually tested
Tested manually against my environment where i encountered the problem and as i use the configurable setting of consumer.pollingTime=1000 ; the error i encounter very often is not reproducible now.

* [ x] Branch build passed

* [ x] No 'package.*' in the imports

* [ ] Relevant Wiki page updated with clear instruction for the end user
  * [ x] Not applicable. This was only a refactor change, no functional or behaviour changes were introduced

* [ ] Http test added to `http-testing` module(if applicable) ?
  * [ x] Not applicable. The changes did not affect HTTP automation flow

* [ ] Kafka test added to `kafka-testing` module(if applicable) ?
  * [x ] 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 98546d86..357bc842 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
@@ -79,20 +79,22 @@ 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;
+    public static ConsumerRecords initialPollWaitingForConsumerGroupJoin(Consumer consumer, ConsumerLocalConfigs effectiveLocalConfigs) {
+
+            for (int run = 0; run < 50; run++) {
+                if (!consumer.assignment().isEmpty()) {
+                    return new ConsumerRecords(new HashMap());
+                }
+                ConsumerRecords records = consumer.poll(Duration.of(getPollTime(effectiveLocalConfigs), ChronoUnit.MILLIS));
+                if (!records.isEmpty()) {
+                    return records;
+                }
             }
-        }
 
-        throw new RuntimeException("\n********* Kafka Consumer unable to join in time *********\n");
+        throw new RuntimeException("\n********* Kafka Consumer unable to join in time - try increasing consumer polling time setting *********\n");
     }
 
+
     public static void validateLocalConfigs(ConsumerLocalConfigs localConfigs) {
         if (localConfigs != null) {
             Boolean localCommitSync = localConfigs.getCommitSync();
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 db64b5f0..9869189e 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
@@ -63,7 +63,7 @@ public class KafkaReceiver {
 
         LOGGER.info("initial polling to trigger ConsumerGroupJoin");
 
-        ConsumerRecords records = initialPollWaitingForConsumerGroupJoin(consumer);
+        ConsumerRecords records = initialPollWaitingForConsumerGroupJoin(consumer, effectiveLocal);
 
         if(!records.isEmpty()) {
             LOGGER.info("Received {} records on initial poll\n", records.count());


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 e7962bc9..fda81e60 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
@@ -160,14 +160,18 @@ public class KafkaConsumerHelperTest {
 
     @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);
+        ConsumerRecords records = initialPollWaitingForConsumerGroupJoin(consumer, consumerEffectiveConfigs);
 
         // then
         assertThat(records.isEmpty(), is(true));
@@ -175,7 +179,11 @@ public class KafkaConsumerHelperTest {
 
     @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>());
 
@@ -185,7 +193,7 @@ public class KafkaConsumerHelperTest {
         Mockito.when(consumerRecords.isEmpty()).thenReturn(false);
 
         // when
-        ConsumerRecords records = initialPollWaitingForConsumerGroupJoin(consumer);
+        ConsumerRecords records = initialPollWaitingForConsumerGroupJoin(consumer, consumerEffectiveConfigs);
 
         // then
         assertThat(records, equalTo(consumerRecords));
@@ -194,7 +202,12 @@ public class KafkaConsumerHelperTest {
 
     @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>());
 
@@ -207,6 +220,6 @@ public class KafkaConsumerHelperTest {
         expectedException.expectMessage("Kafka Consumer unable to join in time");
 
         // when
-        ConsumerRecords records = initialPollWaitingForConsumerGroupJoin(consumer);
+        ConsumerRecords records = initialPollWaitingForConsumerGroupJoin(consumer, consumerEffectiveConfigs);
     }
 }
\ No newline at end of file
diff --git a/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/KafkaConsumePollingTest.java b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/KafkaConsumePollingTest.java
new file mode 100644
index 00000000..0f5395ad
--- /dev/null
+++ b/kafka-testing/src/test/java/org/jsmart/zerocode/integration/tests/kafka/consume/KafkaConsumePollingTest.java
@@ -0,0 +1,25 @@
+package org.jsmart.zerocode.integration.tests.kafka.consume;
+
+import org.jsmart.zerocode.core.domain.Scenario;
+import org.jsmart.zerocode.core.domain.TargetEnv;
+import org.jsmart.zerocode.core.runner.ZeroCodeUnitRunner;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@TargetEnv("kafka_servers/kafka_test_server_polling.properties")
+@RunWith(ZeroCodeUnitRunner.class)
+public class KafkaConsumePollingTest {
+
+    /**
+     * When no polling time is explicitly defined in properties
+     * file e.g consumer.pollingTime
+     * Then intial poll consumer join will default to program
+     * defined default of 500ms.
+     */
+    @Test
+    @Scenario("kafka/consume/test_kafka_consume.json")
+    public void testKafkaConsume() throws Exception {
+    }
+
+}
diff --git a/kafka-testing/src/test/resources/kafka_servers/kafka_test_server_polling.properties b/kafka-testing/src/test/resources/kafka_servers/kafka_test_server_polling.properties
new file mode 100755
index 00000000..623a8c23
--- /dev/null
+++ b/kafka-testing/src/test/resources/kafka_servers/kafka_test_server_polling.properties
@@ -0,0 +1,36 @@
+# =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
+# kafka bootstrap servers comma separated
+# e.g. localhost:9092,host2:9093
+# =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
+kafka.bootstrap.servers=localhost:9092
+
+kafka.producer.properties=kafka_servers/kafka_producer.properties
+kafka.consumer.properties=kafka_servers/kafka_consumer.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
+
+
+# local producer properties
+producer.key1=value1-testv ycvb