This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch v4.0
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
The following commit(s) were added to refs/heads/v4.0 by this push:
new fb4ba1ed [FLINK-39712][Connectors/Kafka] Avoid duplicate
bounded-source completion notifications
fb4ba1ed is described below
commit fb4ba1ed1f55c696b427e7ae8d2ecfe25848d357
Author: Aleksandr Savonin <[email protected]>
AuthorDate: Mon Sep 21 19:12:41 2026 +0200
[FLINK-39712][Connectors/Kafka] Avoid duplicate bounded-source completion
notifications
Backport the bounded-source completion fix to v4.0 from upstream commit
9c69ad689515705361e2a7a3e4d7dd2cccfaf412.
Send completion once per reader registration and defer it until all
clusters finish assignments for reader registration or returned splits.
Re-arm completion when a reader registers again. Adapt the guard to this
release's assignment paths; it has no reported-split recovery gate.
Wire the real completion callback into the test factory and add three
regressions for initial completion, late readers and returned splits.
Assert the exact assigned split set at completion as well as signal counts.
Generated-by: OpenAI Codex (GPT-6)
---
.../enumerator/DynamicKafkaSourceEnumerator.java | 60 ++++++---
.../DynamicKafkaSourceEnumeratorTest.java | 147 ++++++++++++++++++++-
2 files changed, 186 insertions(+), 21 deletions(-)
diff --git
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java
index ff7cc21d..452ad8f5 100644
---
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java
+++
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java
@@ -96,6 +96,8 @@ public class DynamicKafkaSourceEnumerator
private Map<String, Set<String>> latestClusterTopicsMap;
private Set<KafkaStream> latestKafkaStreams;
private boolean firstDiscoveryComplete;
+ private final Set<Integer> readersWithNoMoreSplits = new HashSet<>();
+ private boolean splitAssignmentInProgress;
public DynamicKafkaSourceEnumerator(
KafkaStreamSubscriber kafkaStreamSubscriber,
@@ -212,6 +214,10 @@ public class DynamicKafkaSourceEnumerator
}
private void handleNoMoreSplits() {
+ // A cluster callback may run before other clusters have assigned
their splits.
+ if (splitAssignmentInProgress) {
+ return;
+ }
if (Boundedness.BOUNDED.equals(boundedness)) {
boolean allEnumeratorsHaveSignalledNoMoreSplits = true;
for (StoppableKafkaEnumContextProxy context :
clusterEnumContextMap.values()) {
@@ -220,10 +226,12 @@ public class DynamicKafkaSourceEnumerator
}
if (firstDiscoveryComplete &&
allEnumeratorsHaveSignalledNoMoreSplits) {
- logger.info(
- "Signal no more splits to all readers: {}",
- enumContext.registeredReaders().keySet());
-
enumContext.registeredReaders().keySet().forEach(enumContext::signalNoMoreSplits);
+ for (int readerId : enumContext.registeredReaders().keySet()) {
+ if (readersWithNoMoreSplits.add(readerId)) {
+ logger.info("Signal no more splits to reader {}",
readerId);
+ enumContext.signalNoMoreSplits(readerId);
+ }
+ }
} else {
logger.info("Not ready to notify no more splits to readers.");
}
@@ -456,18 +464,22 @@ public class DynamicKafkaSourceEnumerator
}
// add splits back and assign pending splits for all enumerators
- for (String kafkaClusterId : kafkaPartitionSplits.keySet()) {
- if (clusterEnumeratorMap.containsKey(kafkaClusterId)) {
- clusterEnumeratorMap
- .get(kafkaClusterId)
-
.addSplitsBack(kafkaPartitionSplits.get(kafkaClusterId), subtaskId);
- } else {
- logger.warn(
- "Split refers to inactive cluster {} with current
clusters being {}",
- kafkaClusterId,
- clusterEnumeratorMap.keySet());
- }
- }
+ runWithSplitAssignmentInProgress(
+ () -> {
+ for (String kafkaClusterId :
kafkaPartitionSplits.keySet()) {
+ if (clusterEnumeratorMap.containsKey(kafkaClusterId)) {
+ clusterEnumeratorMap
+ .get(kafkaClusterId)
+ .addSplitsBack(
+
kafkaPartitionSplits.get(kafkaClusterId), subtaskId);
+ } else {
+ logger.warn(
+ "Split refers to inactive cluster {} with
current clusters being {}",
+ kafkaClusterId,
+ clusterEnumeratorMap.keySet());
+ }
+ }
+ });
handleNoMoreSplits();
}
@@ -476,12 +488,24 @@ public class DynamicKafkaSourceEnumerator
@Override
public void addReader(int subtaskId) {
logger.debug("Adding reader {}", subtaskId);
+ readersWithNoMoreSplits.remove(subtaskId);
// assign pending splits from the sub enumerator
- clusterEnumeratorMap.forEach(
- (cluster, subEnumerator) ->
subEnumerator.addReader(subtaskId));
+ runWithSplitAssignmentInProgress(
+ () ->
+ clusterEnumeratorMap.forEach(
+ (cluster, subEnumerator) ->
subEnumerator.addReader(subtaskId)));
handleNoMoreSplits();
}
+ private void runWithSplitAssignmentInProgress(Runnable assignment) {
+ splitAssignmentInProgress = true;
+ try {
+ assignment.run();
+ } finally {
+ splitAssignmentInProgress = false;
+ }
+ }
+
/**
* Besides for checkpointing, this method is used in the restart sequence
to retain the relevant
* assigned splits so that there is no reader duplicate split assignment.
See {@link
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorTest.java
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorTest.java
index 86133345..fe2c64f0 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorTest.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorTest.java
@@ -48,6 +48,7 @@ import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -101,6 +102,81 @@ public class DynamicKafkaSourceEnumeratorTest {
}
}
+ @Test
+ public void testBoundedSourceSignalsNoMoreSplitsOncePerReader() throws
Throwable {
+ try (RecordingSplitEnumeratorContext context = new
RecordingSplitEnumeratorContext();
+ DynamicKafkaSourceEnumerator enumerator =
createBoundedEnumerator(context)) {
+ enumerator.start();
+ for (int reader = 0; reader < NUM_SUBTASKS; reader++) {
+ mockRegisterReaderAndSendReaderStartupEvent(context,
enumerator, reader);
+ }
+
+ assertThat(context.splitsAtCompletion).isEmpty();
+ runAllOneTimeCallables(context);
+
+ for (int reader = 0; reader < NUM_SUBTASKS; reader++) {
+ context.assertReaderCompleted(reader, 1);
+ }
+ verifyAllSplitsHaveBeenAssigned(
+ context.getSplitsAssignmentSequence(),
+ DynamicKafkaSourceTestHelper.getKafkaStream(TOPIC));
+ }
+ }
+
+ @Test
+ public void
testBoundedSourceAssignsAllClustersBeforeCompletingLateReaders() throws
Throwable {
+ try (RecordingSplitEnumeratorContext context = new
RecordingSplitEnumeratorContext();
+ DynamicKafkaSourceEnumerator enumerator =
createBoundedEnumerator(context)) {
+ enumerator.start();
+ mockRegisterReaderAndSendReaderStartupEvent(context, enumerator,
0);
+ runAllOneTimeCallables(context);
+
+ context.assertReaderCompleted(0, 1);
+ assertThat(context.splitsAtCompletion).containsOnlyKeys(0);
+
+ for (int reader = 1; reader < NUM_SUBTASKS; reader++) {
+ mockRegisterReaderAndSendReaderStartupEvent(context,
enumerator, reader);
+ for (int registeredReader = 0; registeredReader <= reader;
registeredReader++) {
+ context.assertReaderCompleted(registeredReader, 1);
+ }
+ }
+ verifyAllSplitsHaveBeenAssigned(
+ context.getSplitsAssignmentSequence(),
+ DynamicKafkaSourceTestHelper.getKafkaStream(TOPIC));
+ }
+ }
+
+ @Test
+ public void testBoundedSourceCompletesRestartedReaderAfterReturnedSplits()
throws Throwable {
+ try (RecordingSplitEnumeratorContext context = new
RecordingSplitEnumeratorContext();
+ DynamicKafkaSourceEnumerator enumerator =
createBoundedEnumerator(context)) {
+ enumerator.start();
+ for (int reader = 0; reader < NUM_SUBTASKS; reader++) {
+ mockRegisterReaderAndSendReaderStartupEvent(context,
enumerator, reader);
+ }
+ runAllOneTimeCallables(context);
+
+ List<DynamicKafkaSourceSplit> returnedSplits =
context.getAssignedSplits(0);
+ context.getSplitsAssignmentSequence().clear();
+ context.unregisterReader(0);
+ enumerator.addSplitsBack(returnedSplits, 0);
+
+ assertThat(context.getSplitsAssignmentSequence()).isEmpty();
+ context.splitsAtCompletion
+ .values()
+ .forEach(completions ->
assertThat(completions).hasSize(1));
+
+ mockRegisterReaderAndSendReaderStartupEvent(context, enumerator,
0);
+
+ assertThat(context.getAssignedSplits(0))
+ .containsExactlyInAnyOrderElementsOf(returnedSplits);
+ context.assertReaderCompleted(0, 2);
+ for (int reader = 1; reader < NUM_SUBTASKS; reader++) {
+ assertThat(context.splitsAtCompletion.get(reader)).hasSize(1);
+ }
+ }
+ }
+
@Test
public void testStartupWithContinuousDiscovery() throws Throwable {
try (MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
@@ -820,6 +896,25 @@ public class DynamicKafkaSourceEnumeratorTest {
}
}
+ private DynamicKafkaSourceEnumerator createBoundedEnumerator(
+ SplitEnumeratorContext<DynamicKafkaSourceSplit> context) {
+ Properties properties = new Properties();
+
properties.setProperty(KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS.key(),
"0");
+ properties.setProperty(
+
DynamicKafkaSourceOptions.STREAM_METADATA_DISCOVERY_INTERVAL_MS.key(), "0");
+ return new DynamicKafkaSourceEnumerator(
+ new KafkaStreamSetSubscriber(Collections.singleton(TOPIC)),
+ new MockKafkaMetadataService(
+
Collections.singleton(DynamicKafkaSourceTestHelper.getKafkaStream(TOPIC))),
+ context,
+ OffsetsInitializer.earliest(),
+ OffsetsInitializer.latest(),
+ properties,
+ Boundedness.BOUNDED,
+ new DynamicKafkaSourceEnumState(),
+ new TestKafkaEnumContextProxyFactory());
+ }
+
private DynamicKafkaSourceEnumerator createEnumerator(
SplitEnumeratorContext<DynamicKafkaSourceSplit> context) {
return createEnumerator(
@@ -1016,6 +1111,50 @@ public class DynamicKafkaSourceEnumeratorTest {
}
}
+ private static class RecordingSplitEnumeratorContext
+ extends MockSplitEnumeratorContext<DynamicKafkaSourceSplit> {
+ private final Map<Integer, List<Set<String>>> splitsAtCompletion = new
HashMap<>();
+
+ private RecordingSplitEnumeratorContext() {
+ super(NUM_SUBTASKS);
+ }
+
+ @Override
+ public void signalNoMoreSplits(int subtask) {
+ super.signalNoMoreSplits(subtask);
+ splitsAtCompletion
+ .computeIfAbsent(subtask, ignored -> new ArrayList<>())
+ .add(
+ getAssignedSplits(subtask).stream()
+ .map(DynamicKafkaSourceSplit::splitId)
+ .collect(Collectors.toSet()));
+ }
+
+ private List<DynamicKafkaSourceSplit> getAssignedSplits(int reader) {
+ return getSplitsAssignmentSequence().stream()
+ .flatMap(
+ assignment ->
+ assignment.assignment()
+ .getOrDefault(reader,
Collections.emptyList()).stream())
+ .collect(Collectors.toList());
+ }
+
+ private void assertReaderCompleted(int reader, int times) {
+ List<DynamicKafkaSourceSplit> assignedSplits =
getAssignedSplits(reader);
+
assertThat(assignedSplits).hasSize(DynamicKafkaSourceTestHelper.NUM_KAFKA_CLUSTERS);
+ Set<String> assignedSplitIds =
+ assignedSplits.stream()
+ .map(DynamicKafkaSourceSplit::splitId)
+ .collect(Collectors.toSet());
+ assertThat(splitsAtCompletion.get(reader))
+ .hasSize(times)
+ .allSatisfy(
+ splitIds ->
+ assertThat(splitIds)
+
.containsExactlyInAnyOrderElementsOf(assignedSplitIds));
+ }
+ }
+
private static class TestKafkaEnumContextProxyFactory
implements
StoppableKafkaEnumContextProxy.StoppableKafkaEnumContextProxyFactory {
@@ -1028,7 +1167,8 @@ public class DynamicKafkaSourceEnumeratorTest {
return new TestKafkaEnumContextProxy(
kafkaClusterId,
kafkaMetadataService,
- (MockSplitEnumeratorContext<DynamicKafkaSourceSplit>)
enumContext);
+ (MockSplitEnumeratorContext<DynamicKafkaSourceSplit>)
enumContext,
+ signalNoMoreSplitsCallback);
}
}
@@ -1039,8 +1179,9 @@ public class DynamicKafkaSourceEnumeratorTest {
public TestKafkaEnumContextProxy(
String kafkaClusterId,
KafkaMetadataService kafkaMetadataService,
- MockSplitEnumeratorContext<DynamicKafkaSourceSplit>
enumContext) {
- super(kafkaClusterId, kafkaMetadataService, enumContext, null);
+ MockSplitEnumeratorContext<DynamicKafkaSourceSplit>
enumContext,
+ Runnable signalNoMoreSplitsCallback) {
+ super(kafkaClusterId, kafkaMetadataService, enumContext,
signalNoMoreSplitsCallback);
this.enumContext = enumContext;
}
