slachiewicz commented on PR #291:
URL:
https://github.com/apache/flink-connector-kafka/pull/291#issuecomment-5471528725
Thanks for the question, @Efrat19!
### Why bounded jobs worked in most cases (and why this went unnoticed)
Bounded jobs were able to complete in most scenarios because this issue
requires three specific conditions to coincide simultaneously:
1. **A bounded stream / batch job** (where a `NoMoreSplitsEvent` is emitted
after split discovery).
2. **`parallelism > total partitions`**, meaning at least one reader subtask
is an **idle reader** receiving 0 splits (`pendingSplits.isEmpty() == true`).
3. **Specific event arrival order**: The idle reader receives
`NoMoreSplitsEvent` (`notifyNoMoreSplits()`) **before** the initial
`MetadataUpdateEvent` is handled.
In normal/happy-path executions:
- **Readers with assigned splits** were never affected because
`!pendingSplits.isEmpty()` evaluated to `true`, which forwarded
`notifyNoMoreSplits()` properly.
- **Idle readers where `MetadataUpdateEvent` arrived first** also completed
normally because `clusterReaderMap` was already populated when
`notifyNoMoreSplits()` was called.
The hang only occurs when an idle reader registers or handles events in the
order `NoMoreSplitsEvent -> MetadataUpdateEvent` (e.g. when an idle subtask
finishes registration slightly after the enumerator has already completed
partition discovery and broadcasted the signal). In that specific order,
`clusterReaderMap` was empty when `notifyNoMoreSplits()` was called, and when
`MetadataUpdateEvent` later populated `clusterReaderMap`, the re-notification
was skipped due to `pendingSplits.isEmpty()`.
### Reproduction in ITCase vs. Unit Test
- In real/integration environments, this depends on non-deterministic
RPC/actor message scheduling. We observed intermittent hangs in
`DynamicKafkaSourceITTest#testIdleReader` in CI under heavy load, which pointed
to this race condition.
- Because reproducing this via full ITCases is subject to thread scheduling
timing, we added the deterministic unit test
`DynamicKafkaSourceReaderTest#testIdleReaderFinishesWhenNoMoreSplitsArrivesBeforeMetadata`.
It isolates this exact sequence:
1. Idle reader starts (0 splits).
2. Receives `notifyNoMoreSplits()`.
3. Receives `MetadataUpdateEvent`.
4. On unfixed `main`, `pollNext()` returns `NOTHING_AVAILABLE`
indefinitely; with this fix, it cleanly emits `END_OF_INPUT`.
### Impact & Safety
The fix is strictly additive and safe:
- `notifyNoMoreSplits()` is idempotent and only executes if `isNoMoreSplits
== true`.
- `MetadataUpdateEvent` is only delivered upon initial reader registration
or actual cluster/topic metadata changes (not on checkpoint intervals), so this
introduces no recurring cross-component overhead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
