This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch v3.4
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
The following commit(s) were added to refs/heads/v3.4 by this push:
new aa7c8ef1 [FLINK-39712][Connectors/Kafka] Avoid duplicate
bounded-source completion notifications
aa7c8ef1 is described below
commit aa7c8ef1388e6cd45706062001f377460561b28d
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 v3.4 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 b61cee40..93a95451 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
@@ -98,6 +98,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,
@@ -214,6 +216,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()) {
@@ -222,10 +228,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.");
}
@@ -463,18 +471,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();
}
@@ -483,12 +495,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;
}
