(flink-connector-kafka) branch main updated: [FLINK-40136][Connectors/Kafka] Fix flaky ConcurrentModificationException in DynamicKafkaSourceEnumeratorTest

[martijnvisser]

This is an automated email from the ASF dual-hosted git repository.

MartijnVisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git


The following commit(s) were added to refs/heads/main by this push:
     new 2354da05 [FLINK-40136][Connectors/Kafka] Fix flaky 
ConcurrentModificationException in DynamicKafkaSourceEnumeratorTest
2354da05 is described below

commit 2354da054b93e422d22d69bb52cef380f39d6684
Author: Sylwester Lachiewicz <slachiew...@apache.org>
AuthorDate: Sun Jul 12 12:48:22 2026 +0200

    [FLINK-40136][Connectors/Kafka] Fix flaky ConcurrentModificationException 
in DynamicKafkaSourceEnumeratorTest
    
    hasLatestMetadataUpdateEvent() polls 
getLatestMetadataUpdateEventWithoutContextSync()
    while the enumerator's metadata discovery thread appends events via
    sendEventToSourceReader() on the main executor thread. In 
MockSplitEnumeratorContext,
    sentSourceEvent's inner List<SourceEvent> instances are shared and can be 
appended
    to concurrently while the polling loop streams them (FLINK-40543), which 
intermittently
    throws ConcurrentModificationException.
    
    Catch ConcurrentModificationException alongside AssertionError in 
hasLatestMetadataUpdateEvent()
    so the existing CommonTestUtils.waitUtil(...) polling loop retries instead 
of failing the test.
    
    Retain reflection in getLatestMetadataUpdateEventWithoutContextSync() 
because calling
    the public MockSplitEnumeratorContext#getSentSourceEvent() dispatches to 
workerExecutor
    and calls Future#get(); in tests that intentionally block workerExecutor 
(such as
    
testProductionMetadataRefreshBypassesBlockedSourceCoordinatorAsyncCallable), 
calling
    getSentSourceEvent() causes a thread deadlock.
    
    Generated-by: Claude Fable 5
---
 .../source/enumerator/DynamicKafkaSourceEnumeratorTest.java   | 11 ++++++++++-
 1 file changed, 10 insertions(+), 1 deletion(-)

diff --git 
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorTest.java
 
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorTest.java
index f75df764..796819c4 100644
--- 
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorTest.java
+++ 
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorTest.java
@@ -61,6 +61,7 @@ import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.ConcurrentModificationException;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -2411,6 +2412,11 @@ public class DynamicKafkaSourceEnumeratorTest {
     @SuppressWarnings("unchecked")
     private MetadataUpdateEvent getLatestMetadataUpdateEventWithoutContextSync(
             MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context, int 
readerId) {
+        // Reflection is required here because 
MockSplitEnumeratorContext#getSentSourceEvent()
+        // dispatches to workerExecutor and blocks on Future#get(). In tests 
that intentionally
+        // block workerExecutor (e.g.
+        // 
testProductionMetadataRefreshBypassesBlockedSourceCoordinatorAsyncCallable),
+        // calling getSentSourceEvent() deadlocks.
         Map<Integer, List<SourceEvent>> sentSourceEvents =
                 (Map<Integer, List<SourceEvent>>)
                         Whitebox.getInternalState(context, "sentSourceEvent");
@@ -2430,6 +2436,9 @@ public class DynamicKafkaSourceEnumeratorTest {
                                                 readerId)));
     }
 
+    // Polling predicate: retries on AssertionError ("not ready yet") as well 
as
+    // ConcurrentModificationException because sentSourceEvent's inner event 
lists are modified
+    // concurrently by the main executor thread (see FLINK-40543).
     private boolean hasLatestMetadataUpdateEvent(
             MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context,
             int readerId,
@@ -2438,7 +2447,7 @@ public class DynamicKafkaSourceEnumeratorTest {
             return getLatestMetadataUpdateEventWithoutContextSync(context, 
readerId)
                     .getKafkaStreams()
                     .equals(Collections.singleton(expectedKafkaStream));
-        } catch (AssertionError e) {
+        } catch (AssertionError | ConcurrentModificationException e) {
             return false;
         }
     }