This is an automated email from the ASF dual-hosted git repository.
bowenli86 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
The following commit(s) were added to refs/heads/main by this push:
new aa6be4ba [FLINK-40094][Connectors/Kafka] Mark removed
DynamicKafkaSource split outputs idle before release (#283)
aa6be4ba is described below
commit aa6be4bac8aeab15b471820f037fcf3cecca044c
Author: lnbest0707 <[email protected]>
AuthorDate: Tue Jul 14 11:16:21 2026 -0700
[FLINK-40094][Connectors/Kafka] Mark removed DynamicKafkaSource split
outputs idle before release (#283)
This implements FLINK-40094 as the mandatory correctness follow-up to
FLINK-39979 and #276.
#276 correctly marks the DynamicKafkaSource reader's main output idle when
metadata removal leaves a source subtask without active subreaders. Flink also
maintains split-local outputs for source watermark aggregation, however, and
directly calling releaseOutputForSplit does not recompute or publish their
aggregate watermark/idleness. A removed split that was active can therefore
leave the subtask output active and pin the downstream watermark even after the
main reader output becomes idle.
This change marks each removed split output idle immediately before
releasing it. That forces Flink to publish the updated aggregate state before
the split-local output is unregistered.
---
.../source/reader/DynamicKafkaSourceReader.java | 17 +++-
.../DynamicKafkaSourceReaderIdlenessTest.java | 100 +++++++++++++++++----
2 files changed, 99 insertions(+), 18 deletions(-)
diff --git
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java
index 3d4dd040..3f4d0469 100644
---
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java
+++
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java
@@ -389,13 +389,19 @@ public class DynamicKafkaSourceReader<T> implements
SourceReader<T, DynamicKafka
notifyNoMoreSplits();
}
}
+
+ // Releasing the last split output can expose the runtime output as
idle immediately. Keep
+ // the reader-level state in sync so a replacement split is
reactivated before polling it.
+ if (latestReaderOutput != null) {
+ maybeUpdateNoActiveSplitOutputIdleness(latestReaderOutput);
+ }
}
private void releaseOrDeferSplitOutput(String splitId) {
if (latestReaderOutput == null) {
pendingSplitOutputReleases.add(splitId);
} else {
- latestReaderOutput.releaseOutputForSplit(splitId);
+ markSplitIdleAndReleaseOutput(latestReaderOutput, splitId);
}
}
@@ -405,11 +411,18 @@ public class DynamicKafkaSourceReader<T> implements
SourceReader<T, DynamicKafka
}
for (String splitId : pendingSplitOutputReleases) {
- readerOutput.releaseOutputForSplit(splitId);
+ markSplitIdleAndReleaseOutput(readerOutput, splitId);
}
pendingSplitOutputReleases.clear();
}
+ private void markSplitIdleAndReleaseOutput(ReaderOutput<T> readerOutput,
String splitId) {
+ // Unregistering a split output does not recompute Flink's split-local
watermark or
+ // idleness. Mark it idle first so a removed split cannot keep the
source output active.
+ readerOutput.createOutputForSplit(splitId).markIdle();
+ readerOutput.releaseOutputForSplit(splitId);
+ }
+
private static boolean isSplitForActiveClusters(
DynamicKafkaSourceSplit split, Map<String, Set<String>> metadata) {
return metadata.containsKey(split.getKafkaClusterId())
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderIdlenessTest.java
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderIdlenessTest.java
index 0a34194f..c3efade0 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderIdlenessTest.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderIdlenessTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.connector.kafka.dynamic.source.reader;
import org.apache.flink.api.common.eventtime.Watermark;
+import org.apache.flink.api.connector.source.SourceOutput;
import org.apache.flink.connector.kafka.dynamic.metadata.ClusterMetadata;
import org.apache.flink.connector.kafka.dynamic.metadata.KafkaStream;
import org.apache.flink.connector.kafka.dynamic.source.MetadataUpdateEvent;
@@ -33,6 +34,8 @@ import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.ArrayList;
import java.util.Collections;
@@ -48,8 +51,10 @@ class DynamicKafkaSourceReaderIdlenessTest {
private static final String REMAINING_TOPIC = "remaining-topic";
private static final String KAFKA_CLUSTER_ID = "cluster-1";
- @Test
- void testMetadataRemovalMarksReaderIdleWhenAllActiveSplitsRemoved() throws
Exception {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testMetadataRemovalMarksReaderAndSplitOutputIdle(boolean
outputAvailableBeforeRemoval)
+ throws Exception {
TestingReaderContext context = new TestingReaderContext();
try (DynamicKafkaSourceReader<byte[]> reader =
new DynamicKafkaSourceReader<>(
@@ -59,33 +64,65 @@ class DynamicKafkaSourceReaderIdlenessTest {
reader.start();
reader.handleSourceEvents(getMetadataUpdateEvent());
- DynamicKafkaSourceSplit split =
- new DynamicKafkaSourceSplit(
- KAFKA_CLUSTER_ID,
- new KafkaPartitionSplit(
- new TopicPartition(TOPIC, 0),
- 0L,
- KafkaPartitionSplit.NO_STOPPING_OFFSET));
+ DynamicKafkaSourceSplit split = getSplit(TOPIC);
reader.addSplits(Collections.singletonList(split));
CompletableFuture<Void> availabilityFuture = reader.isAvailable();
assertThat(availabilityFuture).isNotDone();
+ TrackingReaderOutputWithIdleness<byte[]> readerOutput =
+ new TrackingReaderOutputWithIdleness<>();
+ if (outputAvailableBeforeRemoval) {
+
assertThat(reader.pollNext(readerOutput)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+ }
+
reader.handleSourceEvents(getMetadataUpdateEvent(REMAINING_TOPIC));
assertThat(reader.getAvailabilityHelperSize()).isEqualTo(1);
assertThat(availabilityFuture).isDone();
- TrackingReaderOutputWithIdleness<byte[]> readerOutput =
- new TrackingReaderOutputWithIdleness<>();
assertThat(reader.pollNext(readerOutput)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
assertThat(reader.pollNext(readerOutput)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
-
assertThat(readerOutput.releasedSplitIds()).containsExactly(split.splitId());
+ assertThat(readerOutput.splitOutputEvents())
+ .containsExactly("idle:" + split.splitId(), "release:" +
split.splitId());
assertThat(readerOutput.idleCount()).isEqualTo(1);
assertThat(readerOutput.activeCount()).isZero();
assertThat(readerOutput.isIdle()).isTrue();
}
}
+ @Test
+ void testSplitReassignmentBeforeIdlePollReactivatesReaderOutput() throws
Exception {
+ TestingReaderContext context = new TestingReaderContext();
+ try (DynamicKafkaSourceReader<byte[]> reader =
+ new DynamicKafkaSourceReader<>(
+ context,
+
KafkaRecordDeserializationSchema.valueOnly(ByteArrayDeserializer.class),
+ getRequiredProperties())) {
+ reader.start();
+ reader.handleSourceEvents(getMetadataUpdateEvent());
+
+ DynamicKafkaSourceSplit removedSplit = getSplit(TOPIC);
+ reader.addSplits(Collections.singletonList(removedSplit));
+
+ TrackingReaderOutputWithIdleness<byte[]> readerOutput =
+ new TrackingReaderOutputWithIdleness<>();
+
assertThat(reader.pollNext(readerOutput)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+
+ reader.handleSourceEvents(getMetadataUpdateEvent(REMAINING_TOPIC));
+ assertThat(readerOutput.isIdle()).isTrue();
+
+
reader.addSplits(Collections.singletonList(getSplit(REMAINING_TOPIC)));
+
assertThat(reader.pollNext(readerOutput)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+
+ assertThat(readerOutput.splitOutputEvents())
+ .containsExactly(
+ "idle:" + removedSplit.splitId(), "release:" +
removedSplit.splitId());
+ assertThat(readerOutput.idleCount()).isEqualTo(1);
+ assertThat(readerOutput.activeCount()).isEqualTo(1);
+ assertThat(readerOutput.isIdle()).isFalse();
+ }
+ }
+
private static MetadataUpdateEvent getMetadataUpdateEvent() {
return getMetadataUpdateEvent(TOPIC);
}
@@ -102,6 +139,13 @@ class DynamicKafkaSourceReaderIdlenessTest {
Collections.singletonMap(KAFKA_CLUSTER_ID,
clusterMetadata))));
}
+ private static DynamicKafkaSourceSplit getSplit(String topic) {
+ return new DynamicKafkaSourceSplit(
+ KAFKA_CLUSTER_ID,
+ new KafkaPartitionSplit(
+ new TopicPartition(topic, 0), 0L,
KafkaPartitionSplit.NO_STOPPING_OFFSET));
+ }
+
private static Properties getRequiredProperties() {
Properties properties = new Properties();
properties.setProperty(
@@ -117,7 +161,31 @@ class DynamicKafkaSourceReaderIdlenessTest {
private int idleCount;
private int activeCount;
private boolean idle;
- private final List<String> releasedSplitIds = new ArrayList<>();
+ private final List<String> splitOutputEvents = new ArrayList<>();
+
+ @Override
+ public SourceOutput<E> createOutputForSplit(String splitId) {
+ return new SourceOutput<E>() {
+ @Override
+ public void collect(E record) {}
+
+ @Override
+ public void collect(E record, long timestamp) {}
+
+ @Override
+ public void emitWatermark(Watermark watermark) {}
+
+ @Override
+ public void markIdle() {
+ splitOutputEvents.add("idle:" + splitId);
+ // Model the split-output multiplexer publishing aggregate
idleness.
+ idle = true;
+ }
+
+ @Override
+ public void markActive() {}
+ };
+ }
@Override
public void emitWatermark(Watermark watermark) {}
@@ -136,7 +204,7 @@ class DynamicKafkaSourceReaderIdlenessTest {
@Override
public void releaseOutputForSplit(String splitId) {
- releasedSplitIds.add(splitId);
+ splitOutputEvents.add("release:" + splitId);
}
private int idleCount() {
@@ -151,8 +219,8 @@ class DynamicKafkaSourceReaderIdlenessTest {
return idle;
}
- private List<String> releasedSplitIds() {
- return releasedSplitIds;
+ private List<String> splitOutputEvents() {
+ return splitOutputEvents;
}
}
}
