This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch v5.0
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
The following commit(s) were added to refs/heads/v5.0 by this push:
new deaa7fb4 [FLINK-39712][Connectors/Kafka] Avoid duplicate
bounded-source completion notifications
deaa7fb4 is described below
commit deaa7fb47009856c3a7857282fa8442ac723169b
Author: Aleksandr Savonin <[email protected]>
AuthorDate: Mon Sep 21 19:12:42 2026 +0200
[FLINK-39712][Connectors/Kafka] Avoid duplicate bounded-source completion
notifications
Backport the bounded-source completion fix to v5.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 | 90 +++++++-----
.../DynamicKafkaSourceEnumeratorTest.java | 153 ++++++++++++++++++++-
2 files changed, 209 insertions(+), 34 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 73c0a143..5a756659 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
@@ -97,6 +97,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,
@@ -238,6 +240,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()) {
@@ -246,10 +252,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.");
}
@@ -515,30 +523,35 @@ public class DynamicKafkaSourceEnumerator
@Override
public void addSplitsBack(List<DynamicKafkaSourceSplit> splits, int
subtaskId) {
logger.debug("Adding splits back for {}", subtaskId);
- splitAssignmentStrategy.onSplitsBack(splits, subtaskId);
-
- // separate splits by cluster
- Map<String, List<KafkaPartitionSplit>> kafkaPartitionSplits = new
HashMap<>();
- for (DynamicKafkaSourceSplit split : splits) {
- kafkaPartitionSplits
- .computeIfAbsent(split.getKafkaClusterId(), unused -> new
ArrayList<>())
- .add(split.getKafkaPartitionSplit());
- }
-
- // 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(
+ () -> {
+ splitAssignmentStrategy.onSplitsBack(splits, subtaskId);
+
+ // separate splits by cluster
+ Map<String, List<KafkaPartitionSplit>>
kafkaPartitionSplits = new HashMap<>();
+ for (DynamicKafkaSourceSplit split : splits) {
+ kafkaPartitionSplits
+ .computeIfAbsent(
+ split.getKafkaClusterId(), unused ->
new ArrayList<>())
+ .add(split.getKafkaPartitionSplit());
+ }
+ // 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());
+ }
+ }
+ });
handleNoMoreSplits();
}
@@ -546,14 +559,27 @@ public class DynamicKafkaSourceEnumerator
@Override
public void addReader(int subtaskId) {
logger.debug("Adding reader {}", subtaskId);
- splitAssignmentStrategy.onReaderAdded(subtaskId);
-
- // assign pending splits from the sub enumerator
- clusterEnumeratorMap.forEach(
- (cluster, subEnumerator) ->
subEnumerator.addReader(subtaskId));
+ readersWithNoMoreSplits.remove(subtaskId);
+ runWithSplitAssignmentInProgress(
+ () -> {
+ splitAssignmentStrategy.onReaderAdded(subtaskId);
+
+ // assign pending splits from the sub enumerator
+ 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 40e0dd00..a86cb1f1 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
@@ -108,6 +108,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 =
@@ -1332,6 +1407,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(
@@ -1702,7 +1796,8 @@ public class DynamicKafkaSourceEnumeratorTest {
return new TestKafkaEnumContextProxy(
kafkaClusterId,
kafkaMetadataService,
- (MockSplitEnumeratorContext<DynamicKafkaSourceSplit>)
enumContext);
+ (MockSplitEnumeratorContext<DynamicKafkaSourceSplit>)
enumContext,
+ signalNoMoreSplitsCallback);
}
}
@@ -1714,7 +1809,15 @@ public class DynamicKafkaSourceEnumeratorTest {
String kafkaClusterId,
KafkaMetadataService kafkaMetadataService,
MockSplitEnumeratorContext<DynamicKafkaSourceSplit>
enumContext) {
- super(kafkaClusterId, kafkaMetadataService, enumContext, null);
+ this(kafkaClusterId, kafkaMetadataService, enumContext, null);
+ }
+
+ private TestKafkaEnumContextProxy(
+ String kafkaClusterId,
+ KafkaMetadataService kafkaMetadataService,
+ MockSplitEnumeratorContext<DynamicKafkaSourceSplit>
enumContext,
+ Runnable signalNoMoreSplitsCallback) {
+ super(kafkaClusterId, kafkaMetadataService, enumContext,
signalNoMoreSplitsCallback);
this.enumContext = enumContext;
}
@@ -1736,4 +1839,50 @@ public class DynamicKafkaSourceEnumeratorTest {
period);
}
}
+
+ 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));
+ }
+ }
}
