[
https://issues.apache.org/jira/browse/FLINK-40094?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40094:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Priority: Major
> Labels: pull-request-available
> 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)
