try flush API

mas-chen · Jan 16, 2024 · 563f07e · 563f07e
1 parent 3539df6
commit 563f07e
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/...c/test/java/org/apache/flink/streaming/connectors/kafka/DynamicKafkaSourceTestHelper.java b/...c/test/java/org/apache/flink/streaming/connectors/kafka/DynamicKafkaSourceTestHelper.java
@@ -220,9 +220,12 @@ public static <K, V> void produceToKafka(
                 };
         try (KafkaProducer<K, V> producer = new KafkaProducer<>(props)) {
             for (ProducerRecord<K, V> record : records) {
-                producer.send(record, callback).get();
+                producer.send(record, callback);
             }
+
+            producer.flush();
         }
+
         if (sendingError.get() != null) {
             throw sendingError.get();
         }