Check that the iterator has only a single record

apache · Sep 6, 2023 · 2b184da · 2b184da
1 parent 8362d9a
commit 2b184da
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 3 deletions.
diff --git a/...st/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaInternalProducerITCase.java b/...st/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaInternalProducerITCase.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.streaming.connectors.kafka;
 
+import java.util.Iterator;
 import org.apache.flink.streaming.connectors.kafka.internals.FlinkKafkaInternalProducer;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -239,9 +240,11 @@ private void assertRecord(String topicName, String expectedKey, String expectedV
                 records = kafkaConsumer.poll(Duration.ofMillis(10000));
             }
 
-            ConsumerRecord<String, String> record = records.iterator().next();
+            final Iterator<ConsumerRecord<String, String>> it = records.iterator();
+            ConsumerRecord<String, String> record = it.next();
             assertThat(record.key()).isEqualTo(expectedKey);
             assertThat(record.value()).isEqualTo(expectedValue);
+            assertThat(it.hasNext()).isFalse();
         }
     }
 

diff --git a/...afka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java b/...afka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
@@ -18,6 +18,7 @@
 
 package org.apache.flink.streaming.connectors.kafka;
 
+import java.util.Iterator;
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.JobStatus;
@@ -256,7 +257,10 @@ public void run() {
         } while (System.nanoTime() < deadline);
 
         // cancel the job & wait for the job to finish
-        client.cancel(getRunningJobs(client).iterator().next()).get();
+        final Iterator<JobID> it = getRunningJobs(client).iterator();
+        final JobID jobId = it.next();
+        client.cancel(jobId).get();
+        assertThat(it.hasNext()).isFalse();
         runner.join();
 
         final Throwable t = errorRef.get();
@@ -347,7 +351,10 @@ public void run() {
         } while (System.nanoTime() < deadline);
 
         // cancel the job & wait for the job to finish
-        client.cancel(getRunningJobs(client).iterator().next()).get();
+        final Iterator<JobID> it = getRunningJobs(client).iterator();
+        final JobID jobId = it.next();
+        client.cancel(jobId).get();
+        assertThat(it.hasNext()).isFalse();
         runner.join();
 
         final Throwable t = errorRef.get();