[jira] [Created] (FLINK-40094) DynamicKafkaSource should mark removed split outputs idle before release

["Xin Gao (Jira)" <ji...@apache.org>]

Xin Gao created FLINK-40094:
-------------------------------

             Summary: DynamicKafkaSource should mark removed split outputs idle before release
                 Key: FLINK-40094
                 URL: https://issues.apache.org/jira/browse/FLINK-40094
             Project: Flink
          Issue Type: Bug
          Components: Connectors / Kafka
    Affects Versions: kafka-5.0.0
            Reporter: Xin Gao
             Fix For: kafka-5.1.0


This is the mandatory correctness follow-up to [FLINK-39979|https://issues.apache.org/jira/browse/FLINK-39979] and [apache/flink-connector-kafka#276|https://github.com/apache/flink-connector-kafka/pull/276].

FLINK-39979/#276 correctly marks a DynamicKafkaSource reader's main output idle when metadata removal leaves the source subtask without active subreaders. That change is necessary, but it does not update the split-local outputs maintained by Flink's source watermark multiplexer.

When metadata removes a split that previously emitted a watermark, DynamicKafkaSourceReader currently calls ReaderOutput.releaseOutputForSplit directly. Unregistering the split output does not recompute or publish the split-local aggregate watermark/idleness. The source runtime only publishes IDLE after the combination of main and split-local outputs is idle, so the subtask's last published status can remain ACTIVE and pin the downstream watermark even after the main reader output becomes idle.

*Expected Behavior*

Before a removed split output is released, DynamicKafkaSourceReader marks it idle so Flink publishes the updated aggregate state and the removed split cannot hold back downstream watermark progress.

*Actual Behavior*

The reader's main output becomes idle after FLINK-39979, but the split-local aggregate can remain active because releaseOutputForSplit unregisters the split without first publishing its idle transition.

*Proposed Fix*

Mark each removed split output idle immediately before release, for both the immediate release path and the deferred path used when ReaderOutput is not yet available. Add regression coverage that asserts the exact idle-then-release order in both paths.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)