Sylwester Lachiewicz created FLINK-40543:
--------------------------------------------
Summary: Race condition and ConcurrentModificationException in
MockSplitEnumeratorContext.getSentSourceEvent()
Key: FLINK-40543
URL: https://issues.apache.org/jira/browse/FLINK-40543
Project: Flink
Issue Type: Bug
Components: API / Core
Reporter: Sylwester Lachiewicz
In `org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext`:
1. `sendEventToSourceReader(int, SourceEvent)` modifies `sentSourceEvent`:
- Executed on `mainExecutor`:
`sentSourceEvent.computeIfAbsent(subtaskId, k -> new ArrayList<>()).add(event)`
- sentSourceEvent is a plain java.util.HashMap with java.util.ArrayList
values.
2. getSentSourceEvent() copies sentSourceEvent:
- Executed on workerExecutor: workerExecutor.submit(() -> new
HashMap<>(sentSourceEvent)).get()
- workerExecutor and mainExecutor are two distinct single-threaded
executors created in the constructor.
- Consequently, new HashMap<>(sentSourceEvent) executes concurrently
with sentSourceEvent.computeIfAbsent(...) without synchronization, triggering
ConcurrentModificationException.
- Additionally, new HashMap<>(sentSourceEvent) is a shallow copy sharing
the underlying ArrayList<SourceEvent> instances, causing CME if a test thread
streams/iterates the list while mainExecutor appends new events.
Proposed fix:
- Run the map copy on mainExecutor (or use thread-safe data structures like
ConcurrentHashMap).
- Deep-copy the List<SourceEvent> values so callers do not share mutable
ArrayList instances with the writer thread.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
