[
https://issues.apache.org/jira/browse/FLINK-40543?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18114815#comment-18114815
]
Qiu Yanjun commented on FLINK-40543:
------------------------------------
I would like to work on this issue. Could a committer please assign it to me?
> 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
> Priority: Minor
>
> In `org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext`:
> 1. `sendEventToSourceReader(int, SourceEvent)` modifies `sentSourceEvent`:
> - Runs on `mainExecutor` (or current thread if already on the main thread):
> `sentSourceEvent.computeIfAbsent(subtaskId, k -> new
> ArrayList<>()).add(event)`
> - `sentSourceEvent` is a plain `java.util.HashMap` containing standard
> `java.util.ArrayList` instances.
> 2. `getSentSourceEvent()` reads and copies `sentSourceEvent`:
> - Runs 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(...)` on `mainExecutor` without
> synchronization, occasionally throwing `ConcurrentModificationException`
> (wrapped in `ExecutionException`).
> - In addition, `new HashMap<>(sentSourceEvent)` is only a shallow copy:
> the inner `List<SourceEvent>` instances are shared. Callers streaming or
> iterating these lists (e.g. during test assertions or polling) experience
> `ConcurrentModificationException` if `mainExecutor` concurrently appends new
> events.
> ### Recommended Fix:
> - **Do NOT simply delegate the copy to `mainExecutor`**: Unit tests
> frequently block or wait on the coordinator thread (`mainExecutor`) with
> latches to assert intermediate states. Executing
> `mainExecutor.submit(...).get()` inside `getSentSourceEvent()` would cause
> test deadlocks in those scenarios.
> - **Use thread-safe / snapshot-isolated collections**:
> - Store subtask event lists using a concurrent structure, e.g.:
> `ConcurrentMap<Integer, CopyOnWriteArrayList<SourceEvent>>
> sentSourceEvent = new ConcurrentHashMap<>();`
> - Lock-free, wait-free reads; iterators operate on immutable array
> snapshots (completely immune to `ConcurrentModificationException`).
> - $O(N)$ write cost is negligible because unit tests only send a handful
> of events ($N < 100$).
> - Or use `ConcurrentHashMap<Integer, List<SourceEvent>>` and synchronize on
> the individual subtask list when taking an `ArrayList` snapshot during
> `getSentSourceEvent()`.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
