[
https://issues.apache.org/jira/browse/FLINK-40586?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18119501#comment-18119501
]
Bhanu Chander Vallabaneni commented on FLINK-40586:
---------------------------------------------------
PR [#309|https://github.com/apache/flink-connector-kafka/pull/309] has been
rebased and re-scoped, so recording where this ticket now stands.
[~martijnvisser] found that
[#300|https://github.com/apache/flink-connector-kafka/pull/300] (FLINK-39712,
open since 7 September) already carried the per-reader dedup this ticket
describes, and it merged on 21 September. So the fix for the behaviour in the
summary is on {{main}}, contributed there rather than here, and #309 is reduced
to the one thing #300 does not cover: clearing the set when a metadata change
recreates the sub enumerators.
h3. Why the dedup could not land on its own
{{handleNoMoreSplits}} is reached from the proxy callback part way through
{{addReaderToClusterEnumerators}}, so the first signal goes out before the
other clusters have assigned. On {{main}} a later signal corrects it; once the
set makes that the only signal, the reader finishes holding one cluster's
splits. Four of #300's five tests fail against the original #309 branch for
exactly that reason, which I confirmed rather than taking on trust. #300's
deferral is what makes the dedup safe.
h3. The remaining gap, and a finding worth recording
A metadata change closes and recreates every sub enumerator and the reader
recreates its sub readers, so every sub reader is back to
{{noMoreSplitsAssignment == false}}. Entries left in the dedup set from the
previous generation then suppress the signals the new generation needs, and a
bounded job does not finish. #309 now clears the set at
{{closeAllEnumeratorsAndContexts()}}.
While writing the test for it I found why this is hard to observe, which is
relevant to FLINK-31006:
{code:java}
// KafkaSourceEnumerator#checkPartitionChanges
final PartitionChange partitionChange = getPartitionChange(fetchedPartitions,
!initialDiscoveryFinished);
if (partitionChange.isEmpty()) {
return;
}
{code}
A *retained* cluster is recreated from state in which its partitions are
already assigned, so the change is empty and it returns before reaching
{{signalNoMoreSplits}}. Its new proxy's {{isNoMoreSplits()}} therefore stays
false, {{allEnumeratorsHaveSignalledNoMoreSplits}} is never true, and
{{handleNoMoreSplits}} takes the "not ready" branch regardless of the set's
contents. Instrumented:
{code}
contexts=[kafka-cluster-0, kafka-cluster-1]
noMoreSplitsFlags=[kafka-cluster-0=false, kafka-cluster-1=true]
alreadySignalled=[]
{code}
The test therefore drives a cluster *switchover*, where no sub enumerator is
retained, every one is new, all of them signal, and the precondition is
reachable. It fails on {{main}} with the readers signalled once instead of
twice.
h3. This summary is now stale
The summary still describes the re-signalling that #300 fixed. [~martijnvisser]
mentioned he would re-scope the ticket; I have left it alone rather than
editing it myself, but something like "DynamicKafkaSourceEnumerator does not
clear its no-more-splits dedup set when a metadata change recreates the sub
enumerators, so a bounded job does not finish" would match what is left. Happy
for it to be worded however you prefer.
Local verification on the rebased tree: {{DynamicKafkaSourceEnumeratorTest}}
36/36, {{DynamicKafkaSourceEnumeratorRecoveryTest}} 10/10,
{{DynamicKafkaSourceReaderTest}} 15/15. {{spotless:check}} and
{{checkstyle:check}} clean on JDK 17. The workflow run on the PR is at
{{action_required}} after the force-push, so the JDK 11/17/21 legs need a
committer to approve them.
> 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
> Assignee: Bhanu Chander Vallabaneni
> Priority: Major
> Labels: pull-request-available
>
> 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)
