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)
