[
https://issues.apache.org/jira/browse/FLINK-40543?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Sylwester Lachiewicz updated FLINK-40543:
-----------------------------------------
Description:
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()`.
was:
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.
> 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)
