This is an automated email from the ASF dual-hosted git repository.
MartijnVisser 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 9c69ad68 [FLINK-39712][Connectors/Kafka] Avoid duplicate
bounded-source completion notifications
9c69ad68 is described below
commit 9c69ad689515705361e2a7a3e4d7dd2cccfaf412
Author: Aleksandr Savonin <[email protected]>
AuthorDate: Mon Sep 7 12:06:19 2026 +0200
[FLINK-39712][Connectors/Kafka] Avoid duplicate bounded-source completion
notifications
A bounded DynamicKafkaSource can signal readers again after they have
finished when another reader registers or returns splits. An undeliverable
NoMoreSplitsEvent then fails the job. Record which readers have received
completion and re-arm the signal when a reader registers again.
Deduplication must be paired with assignment deferral: a cluster callback
can report completion before the remaining clusters assign their splits.
Defer completion across returned-split assignment, reader registration and
reported-split recovery, then recheck completion after each operation.
Keep the guard active through the recovery-gate drain because draining
clears its pending state before active and retained split assignment ends.
Centralize that synchronous guard in a private Runnable helper that clears
the flag in finally while preserving each completion callback's ordering.
Wire the real completion callback into the enumerator test context and add
five regressions covering startup, late registration, returned splits,
reported splits, and active plus retained reassignment. Assert both signal
counts and the exact split set present at completion.
Generated-by: OpenAI Codex (GPT-6)
---
.../enumerator/DynamicKafkaSourceEnumerator.java | 51 +++--
.../DynamicKafkaSourceEnumeratorTest.java | 230 ++++++++++++++++++++-
2 files changed, 266 insertions(+), 15 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 6cc98d91..60a2fe9b 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
@@ -115,6 +115,8 @@ public class DynamicKafkaSourceEnumerator
retainedClusterEnumeratorStates;
private boolean firstDiscoveryComplete;
private final ReaderRecoveryGate readerRecoveryGate;
+ private final Set<Integer> readersWithNoMoreSplits = new HashSet<>();
+ private boolean splitAssignmentInProgress;
public DynamicKafkaSourceEnumerator(
KafkaStreamSubscriber kafkaStreamSubscriber,
@@ -410,6 +412,10 @@ public class DynamicKafkaSourceEnumerator
}
private void handleNoMoreSplits() {
+ // A cluster callback may run before other clusters have assigned
their splits.
+ if (splitAssignmentInProgress ||
readerRecoveryGate.hasPendingRecovery()) {
+ return;
+ }
if (Boundedness.BOUNDED.equals(boundedness)) {
boolean allEnumeratorsHaveSignalledNoMoreSplits = true;
for (StoppableKafkaEnumContextProxy context :
clusterEnumContextMap.values()) {
@@ -418,10 +424,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.");
}
@@ -751,8 +759,11 @@ public class DynamicKafkaSourceEnumerator
@Override
public void addSplitsBack(List<DynamicKafkaSourceSplit> splits, int
subtaskId) {
logger.debug("Adding splits back for {}", subtaskId);
- splitAssignmentStrategy.onSplitsBack(splits, subtaskId);
- addSplitsBackToClusterEnumerators(splits, subtaskId, false);
+ runWithSplitAssignmentInProgress(
+ () -> {
+ splitAssignmentStrategy.onSplitsBack(splits, subtaskId);
+ addSplitsBackToClusterEnumerators(splits, subtaskId,
false);
+ });
handleNoMoreSplits();
}
@@ -791,6 +802,7 @@ public class DynamicKafkaSourceEnumerator
@Override
public void addReader(int subtaskId) {
logger.debug("Adding reader {}", subtaskId);
+ readersWithNoMoreSplits.remove(subtaskId);
ReaderInfo readerInfo = enumContext.registeredReaders().get(subtaskId);
if (readerInfo != null) {
readerRecoveryGate.recordReportedSplits(
@@ -801,7 +813,7 @@ public class DynamicKafkaSourceEnumerator
return;
}
- addReaderToClusterEnumerators(subtaskId);
+ runWithSplitAssignmentInProgress(() ->
addReaderToClusterEnumerators(subtaskId));
handleNoMoreSplits();
}
@@ -813,17 +825,30 @@ public class DynamicKafkaSourceEnumerator
return true;
}
- readerRecoveryGate.markInitialRegistrationComplete();
- if (readerRecoveryGate.hasReportedSplits()) {
- reassignReportedSplits();
- } else {
- flushPendingSplitAssignmentsForRegisteredReaders();
- }
+ // Draining the gate clears its pending state before reassignment
finishes.
+ runWithSplitAssignmentInProgress(
+ () -> {
+ readerRecoveryGate.markInitialRegistrationComplete();
+ if (readerRecoveryGate.hasReportedSplits()) {
+ reassignReportedSplits();
+ } else {
+ flushPendingSplitAssignmentsForRegisteredReaders();
+ }
+ });
handleNoMoreSplits();
flushPendingMetadataUpdateEvents();
return true;
}
+ private void runWithSplitAssignmentInProgress(Runnable assignment) {
+ splitAssignmentInProgress = true;
+ try {
+ assignment.run();
+ } finally {
+ splitAssignmentInProgress = false;
+ }
+ }
+
private boolean allReadersRegistered() {
return enumContext.registeredReaders().size() ==
enumContext.currentParallelism();
}
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 796819c4..389beb68 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
@@ -127,6 +127,158 @@ 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
testBoundedSourceReassignsReportedSplitsBeforeCompletingReader() 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> reportedSplits =
context.getAssignedSplits(0);
+ context.getSplitsAssignmentSequence().clear();
+ context.unregisterReader(0);
+ context.registerReader(ReaderInfo.createReaderInfo(0, "restarted",
reportedSplits));
+ enumerator.addReader(0);
+
+ assertThat(context.getAssignedSplits(0))
+ .containsExactlyInAnyOrderElementsOf(reportedSplits);
+ context.assertReaderCompleted(0, 2);
+ for (int reader = 1; reader < NUM_SUBTASKS; reader++) {
+ assertThat(context.splitsAtCompletion.get(reader)).hasSize(1);
+ }
+ }
+ }
+
+ @Test
+ public void
testBoundedSourceCompletesReadersAfterActiveAndRetainedReassignment()
+ 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);
+
+ Map<Integer, List<DynamicKafkaSourceSplit>> reportedSplitsByReader
= new HashMap<>();
+ for (int reader = 0; reader < 2; reader++) {
+ reportedSplitsByReader.put(reader,
context.getAssignedSplits(reader));
+ context.unregisterReader(reader);
+ }
+ reportedSplitsByReader
+ .get(0)
+ .add(
+ new DynamicKafkaSourceSplit(
+ "removed-cluster",
+ new KafkaPartitionSplit(
+ new
TopicPartition("removed-topic", 0), 5),
+ Long.MAX_VALUE));
+ context.getSplitsAssignmentSequence().clear();
+ context.splitsAtCompletion.clear();
+
+ context.registerReader(
+ ReaderInfo.createReaderInfo(0, "restarted",
reportedSplitsByReader.get(0)));
+ enumerator.addReader(0);
+ assertThat(context.getSplitsAssignmentSequence()).isEmpty();
+ assertThat(context.splitsAtCompletion).isEmpty();
+
+ context.registerReader(
+ ReaderInfo.createReaderInfo(1, "restarted",
reportedSplitsByReader.get(1)));
+ enumerator.addReader(1);
+
+ assertThat(context.splitsAtCompletion).containsOnlyKeys(0, 1);
+ for (int reader = 0; reader < 2; reader++) {
+ List<DynamicKafkaSourceSplit> reportedSplits =
reportedSplitsByReader.get(reader);
+ assertThat(context.getAssignedSplits(reader))
+ .containsExactlyInAnyOrderElementsOf(reportedSplits);
+ assertThat(context.splitsAtCompletion.get(reader))
+ .singleElement()
+ .isEqualTo(
+ reportedSplits.stream()
+ .map(DynamicKafkaSourceSplit::splitId)
+ .collect(Collectors.toSet()));
+ }
+ }
+ }
+
@Test
public void testStartupWithContinuousDiscovery() throws Throwable {
try (MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
@@ -1999,6 +2151,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(
@@ -2476,7 +2647,8 @@ public class DynamicKafkaSourceEnumeratorTest {
return new TestKafkaEnumContextProxy(
kafkaClusterId,
kafkaMetadataService,
- (MockSplitEnumeratorContext<DynamicKafkaSourceSplit>)
enumContext);
+ (MockSplitEnumeratorContext<DynamicKafkaSourceSplit>)
enumContext,
+ signalNoMoreSplitsCallback);
}
}
@@ -2488,7 +2660,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;
}
@@ -2605,6 +2785,52 @@ 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 DroppingCoordinatorThreadContext
extends MockSplitEnumeratorContext<DynamicKafkaSourceSplit> {
private DroppingCoordinatorThreadContext(int parallelism) {
