Workaround for tests failing due to behavior change in first consumer poll

Author: jzakaryan

datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/AbstractKafkaBasedConnectorTask.java

@@ -508,7 +508,13 @@ protected void trackEventsProcessedProgress(int recordCount) {
   }
 
   protected ConsumerRecords<?, ?> consumerPoll(long pollInterval) {
-    return _consumer.poll(Duration.ofMillis(pollInterval));
+    if (pollInterval == 0) {
+      // Brooklin calls poll with 0 pollInterval when a task is newly initialized. There's a behavior change between old
+      // and new poll in this case. We need to understand that behavior better before removing usages of deprecated API
+      return _consumer.poll(pollInterval);
+    } else {
+      return _consumer.poll(Duration.ofMillis(pollInterval));
+    }
   }
 
   /**

datastream-kafka-connector/src/main/java/com/linkedin/datastream/connectors/kafka/mirrormaker/KafkaMirrorMakerConnectorTask.java

@@ -423,6 +423,10 @@ protected void maybeCommitOffsets(Consumer<?, ?> consumer, boolean hardCommit) {
     if (_enablePartitionAssignment && _consumerAssignment.isEmpty()) {
       // Kafka rejects a poll if there is empty assignment
       return ConsumerRecords.EMPTY;
+    } else if (pollInterval == 0) {
+      // BMM calls poll with 0 pollInterval when a task is newly initialized. There's a behavior change between old poll
+      // and new poll in this case. We need to understand that behavior better before removing usages of deprecated API
+      return _consumer.poll(pollInterval);
     } else {
       return _consumer.poll(Duration.ofMillis(pollInterval));
     }

gradle/maven.gradle

@@ -1,5 +1,5 @@
 allprojects {
-  version = "5.3.0-SNAPSHOT"
+  version = "5.3.2-SNAPSHOT"
 }
 
 subprojects {