[
https://issues.apache.org/jira/browse/FLINK-40586?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18113609#comment-18113609
]
Bhanu Chander Vallabaneni commented on FLINK-40586:
---------------------------------------------------
Confirmed the root cause as described, on {{main}}.
{{DynamicKafkaSourceEnumerator.handleNoMoreSplits()}} signals every currently
registered reader on every invocation, with no record of which readers were
already signalled:
{code:java}
enumContext.registeredReaders().keySet().forEach(enumContext::signalNoMoreSplits);
{code}
and it is reached from three places -- {{addReader}}, {{addSplitsBack}} and
{{tryCompletePendingReaderRegistration}}. With parallelism greater than the
split count, the idle reader finishes on the first signal, and the next reader
registration re-sends {{NoMoreSplitsEvent}} to a task already in {{FINISHED}}.
The event is not loss-tolerant, so the coordinator fails the job, which is the
{{OperatorEvent ... was lost}} failure in the report.
Worth noting why this has no unit coverage today:
{{DynamicKafkaSourceEnumeratorTest.TestKafkaEnumContextProxyFactory}} passes
{{null}} as the {{signalNoMoreSplitsCallback}}, so {{handleNoMoreSplits}} is
never invoked from the proxy in those tests. {{MockSplitEnumeratorContext}}
also only records a {{boolean[] subtaskHasNoMoreSplits}} and exposes
{{hasNoMoreSplits(int)}}, so a repeated signal is invisible to it -- catching
this needs a context that counts calls rather than latching a flag.
The fix I would propose is to remember which subtasks have been signalled and
only signal the ones that have not, clearing a subtask's entry in
{{addSplitsBack}} since splits returning means that reader is restarting and
will re-register. One thing I would like a second opinion on: when splits come
back there are by definition more splits to hand out, so arguably
{{addSplitsBack}} should not be reaching the signalling path at all until the
next discovery settles -- happy to follow whichever reading you prefer.
This is distinct from FLINK-40362, which is the reader-side hang and already
has PRs from [~slachiewicz].
Could someone assign this to me? I am happy to put up the fix with a test that
drives two reader registrations and asserts the first reader is signalled
exactly once.
> DynamicKafkaSourceEnumerator re-signals NoMoreSplits to already-finished
> readers on every reader registration, causing "OperatorEvent lost" task
> failover
> ---------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-40586
> URL: https://issues.apache.org/jira/browse/FLINK-40586
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
> Affects Versions: kafka-3.4.0, kafka-4.0.1, kafka-5.0.0, kafka-5.1.0
> Reporter: Martijn Visser
> Priority: Major
>
> DynamicKafkaSourceITTest$IntegrationTests.testIdleReader fails in the weekly
> CI with
> org.apache.flink.util.FlinkException: An OperatorEvent from an
> OperatorCoordinator to a task was
> lost. Triggering task failover to ensure consistency. Event:
> '[NoMoreSplitEvent]', targetTask:
> Source: Tested Source (3/5) - execution #0
> Caused by:
> org.apache.flink.runtime.operators.coordination.TaskNotRunningException: Task
> is not
> running, but in state FINISHED
> e.g. run 33283936010 job 99183711637 (2026-08-30, main, JDK 11) and run
> 29667371405 job 88140088972 (2026-07-19, v5.0, Flink 2.1.2, JDK 21); also
> seen 2026-07-05 (v4.0) and 2026-08-02 (v5.0/2.2.1).
> DynamicKafkaSourceEnumerator.handleNoMoreSplits() calls
> enumContext.signalNoMoreSplits for every registered reader on every
> invocation, and it is invoked from addReader, addSplitsBack and
> tryCompletePendingReaderRegistration without remembering which readers were
> already signalled. testIdleReader uses parallelism = number of splits + 1, so
> the idle reader finishes as soon as it receives the first signal; the next
> reader registration re-sends NoMoreSplitsEvent to the FINISHED task, the
> event is not loss-tolerant, and the coordinator fails the job.
> This is distinct from FLINK-40362, which fixes the reader-side hang where an
> idle reader swallows the signal when it arrives before the
> MetadataUpdateEvent. That fix makes idle readers finish earlier and, if
> anything, widens this window.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
