vbhanuchander-lang opened a new pull request, #309: URL: https://github.com/apache/flink-connector-kafka/pull/309
## What is the purpose of the change Fixes [FLINK-40586](https://issues.apache.org/jira/browse/FLINK-40586). `DynamicKafkaSourceEnumerator.handleNoMoreSplits()` signalled **every** currently registered reader on **every** invocation, with no record of which readers had already been signalled: ```java enumContext.registeredReaders().keySet().forEach(enumContext::signalNoMoreSplits); ``` It is reached from three places — `addReader`, `addSplitsBack` and `tryCompletePendingReaderRegistration` — so a reader that finished on an earlier signal is signalled again when another reader registers. `NoMoreSplitsEvent` is not loss tolerant, so delivering it to a task already in `FINISHED` fails the job: ``` org.apache.flink.util.FlinkException: An OperatorEvent from an OperatorCoordinator to a task was lost. Triggering task failover to ensure consistency. Event: '[NoMoreSplitEvent]' Caused by: TaskNotRunningException: Task is not running, but in state FINISHED ``` That is the `DynamicKafkaSourceITTest$IntegrationTests.testIdleReader` failure seen in weekly CI, where parallelism is one greater than the split count so the idle reader finishes on the first signal. ## Brief change log - `handleNoMoreSplits()` signals only registered readers that have not been signalled yet, and records them in `readersSignalledNoMoreSplits`. - `addReader()` clears the reader's entry, so each reader is signalled once **per registration** and a reader that re-registers after failover is signalled again. Keying the reset on registration rather than on `addSplitsBack` is deliberate. An idle reader has no splits to hand back, so `addSplitsBack` is never called for it; resetting there would leave a restarted idle reader permanently un-signalled and the job would never finish. ## Verifying this change Added `DynamicKafkaSourceEnumeratorTest#testNoMoreSplitsIsSignalledOncePerReaderRegistration`, which registers three readers and asserts each is signalled exactly once, then hands one reader's splits back to drive `handleNoMoreSplits()` again and asserts the others are not re-signalled. `MockSplitEnumeratorContext` only latches a `boolean[] subtaskHasNoMoreSplits` and exposes `hasNoMoreSplits(int)`, so a repeated signal is invisible to it. The test therefore uses a small `CountingSignalNoMoreSplitsContext` subclass that counts calls per subtask. Against unfixed `main` the new test fails, with reader 0 signalled three times — once per subsequent registration: ``` expected: 1 but was: 3 ``` With the change, the full test class passes: ``` Tests run: 31, Failures: 0, Errors: 0, Skipped: 0 -- DynamicKafkaSourceEnumeratorTest ``` `spotless:check` and `checkstyle:check` pass. Worth noting this is distinct from FLINK-40362, which is the reader-side hang and has separate PRs open. ## Does this pull request potentially affect one of the following parts - Dependencies (does it add or upgrade a dependency): **no** - The public API: **no** - The serializers: **no** - The runtime per-record code paths: **no** - Anything that affects deployment or recovery: **yes** — it changes when `NoMoreSplitsEvent` is delivered to readers on registration and failover, for bounded sources only. - The S3 file system connector: **no** ## Documentation - Does this pull request introduce a new feature? **no** - If yes, how is the feature documented? **not applicable** -- 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]
