[FLINK-30056][kafka] Use new Consumer#poll(Duration)

This changes the blocking behavior, as the previous method essentially ignored the timeout.
apache · Nov 25, 2022 · 5e0e0fd · 5e0e0fd
1 parent 9d87d33
commit 5e0e0fd
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 2 deletions.
diff --git a/.../main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaConsumerThread.java b/.../main/java/org/apache/flink/streaming/connectors/kafka/internals/KafkaConsumerThread.java
@@ -36,6 +36,7 @@
 
 import javax.annotation.Nonnull;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -258,7 +259,7 @@ public void run() {
                 // over
                 if (records == null) {
                     try {
-                        records = consumer.poll(pollTimeout);
+                        records = consumer.poll(Duration.ofMillis(pollTimeout));
                     } catch (WakeupException we) {
                         continue;
                     }

diff --git a/...st/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaInternalProducerITCase.java b/...st/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaInternalProducerITCase.java
@@ -238,7 +238,7 @@ private void assertRecord(String topicName, String expectedKey, String expectedV
             kafkaConsumer.subscribe(Collections.singletonList(topicName));
             ConsumerRecords<String, String> records = ConsumerRecords.empty();
             while (records.isEmpty()) {
-                records = kafkaConsumer.poll(10000);
+                records = kafkaConsumer.poll(Duration.ofMillis(10000));
             }
 
             ConsumerRecord<String, String> record = Iterables.getOnlyElement(records);