[FLINK-30056][Connectors/Kafka] Make polling for metadata no more tha…

…n specified timeout by using new Consumer#poll(Duration)
apache · Nov 23, 2022 · 22e4694 · 22e4694
1 parent 7828d77
commit 22e4694
Show file tree

Hide file tree

Showing 2 changed files with 2 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
@@ -259,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);