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 bb1ed100 [FLINK-41088][Connectors/Kafka] Key recovery reports by
partition
bb1ed100 is described below
commit bb1ed100d698e6205aee8818c5e2cd14fee3ae14
Author: MartijnVisser <[email protected]>
AuthorDate: Thu Oct 22 20:29:48 2026 +0200
[FLINK-41088][Connectors/Kafka] Key recovery reports by partition
The formatted split id is <cluster>-<topic>-<partition>, so cluster a
with topic b-c and cluster a-b with topic c share the id a-b-c-0.
ReaderRecoveryGate merged a reader's reports by that id and dropped one
of the two partitions. reassignReportedSplits rejected two active ones
reported by different readers, which failed every restore.
Both now key by cluster id and TopicPartition, as the restarted-reader
path already does. A partition reported by two readers still fails the
restore, and the message now names the cluster.
Generated-by: Claude Code 2.1.308 (Claude Opus 5.5)
---
.../enumerator/DynamicKafkaSourceEnumerator.java | 25 +-
.../source/enumerator/ReaderRecoveryGate.java | 15 +-
.../DynamicKafkaSourceEnumeratorRecoveryTest.java | 282 +++++++++++++++++++--
.../source/enumerator/ReaderRecoveryGateTest.java | 23 ++
4 files changed, 315 insertions(+), 30 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 487a7400..19d580dc 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
@@ -83,6 +83,11 @@ public class DynamicKafkaSourceEnumerator
private static final Logger logger =
LoggerFactory.getLogger(DynamicKafkaSourceEnumerator.class);
+ /** Deterministic order for splits whose formatted ids can collide across
clusters. */
+ private static final Comparator<DynamicKafkaSourceSplit> SPLIT_ORDER =
+ Comparator.comparing(DynamicKafkaSourceSplit::splitId)
+ .thenComparing(DynamicKafkaSourceSplit::getKafkaClusterId);
+
// Each cluster will have its own sub enumerator
private final Map<String, SplitEnumerator<KafkaPartitionSplit,
KafkaSourceEnumState>>
clusterEnumeratorMap;
@@ -930,7 +935,7 @@ public class DynamicKafkaSourceEnumerator
}
private void reassignReportedSplits() {
- Map<String, ReportedSplit> activeReportedSplits = new TreeMap<>();
+ Map<Tuple2<String, TopicPartition>, ReportedSplit>
activeReportedSplits = new HashMap<>();
Map<Integer, List<DynamicKafkaSourceSplit>> retainedSplitsByReader =
new TreeMap<>();
long currentTimeMillis = System.currentTimeMillis();
@@ -942,13 +947,13 @@ public class DynamicKafkaSourceEnumerator
DynamicKafkaSourceSplit activeSplit =
split.clearRetention();
ReportedSplit previous =
activeReportedSplits.putIfAbsent(
- activeSplit.splitId(),
+ toPartitionKey(activeSplit),
new ReportedSplit(activeSplit, readerId));
if (previous != null) {
throw new IllegalStateException(
String.format(
"Split %s was reported by both reader
%d and reader %d",
- activeSplit.splitId(),
previous.readerId, readerId));
+ activeSplit, previous.readerId,
readerId));
}
} else {
DynamicKafkaSourceSplit retainedSplit =
@@ -964,8 +969,11 @@ public class DynamicKafkaSourceEnumerator
}
}
+ List<ReportedSplit> orderedReportedSplits = new
ArrayList<>(activeReportedSplits.values());
+ orderedReportedSplits.sort(
+ Comparator.comparing(reportedSplit -> reportedSplit.split,
SPLIT_ORDER));
List<DynamicKafkaSourceSplit> activeSplits =
- activeReportedSplits.values().stream()
+ orderedReportedSplits.stream()
.map(reportedSplit -> reportedSplit.split)
.collect(Collectors.toList());
splitAssignmentStrategy.onRecoveredSplits(activeSplits,
enumContext.currentParallelism());
@@ -975,7 +983,7 @@ public class DynamicKafkaSourceEnumerator
// With the initial gate closed, registeredReaders() is empty, so
the reader ID
// has no effect here. The flush after reconciliation covers all
readers. Enqueue
// one batch per cluster to avoid scanning pending copies once per
reported split.
- int reportingReader =
activeReportedSplits.values().iterator().next().readerId;
+ int reportingReader = orderedReportedSplits.get(0).readerId;
addSplitsBackToClusterEnumerators(activeSplits, reportingReader,
true);
}
@@ -995,9 +1003,7 @@ public class DynamicKafkaSourceEnumerator
.drainReportedSplits(readerId)
.forEach(split -> reportedSplits.put(toPartitionKey(split),
split));
List<DynamicKafkaSourceSplit> orderedSplits = new
ArrayList<>(reportedSplits.values());
- orderedSplits.sort(
- Comparator.comparing(DynamicKafkaSourceSplit::splitId)
-
.thenComparing(DynamicKafkaSourceSplit::getKafkaClusterId));
+ orderedSplits.sort(SPLIT_ORDER);
for (DynamicKafkaSourceSplit split : orderedSplits) {
if (!isCurrentSplitOwner(split, readerId)) {
continue;
@@ -1278,7 +1284,8 @@ public class DynamicKafkaSourceEnumerator
}
}
- private static Tuple2<String, TopicPartition>
toPartitionKey(DynamicKafkaSourceSplit split) {
+ /** Identifies the physical partition; formatted split ids collide across
clusters. */
+ static Tuple2<String, TopicPartition>
toPartitionKey(DynamicKafkaSourceSplit split) {
return Tuple2.of(
split.getKafkaClusterId(),
split.getKafkaPartitionSplit().getTopicPartition());
}
diff --git
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/ReaderRecoveryGate.java
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/ReaderRecoveryGate.java
index 4cc8656e..b4e34166 100644
---
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/ReaderRecoveryGate.java
+++
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/ReaderRecoveryGate.java
@@ -19,8 +19,11 @@
package org.apache.flink.connector.kafka.dynamic.source.enumerator;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.java.tuple.Tuple2;
import
org.apache.flink.connector.kafka.dynamic.source.split.DynamicKafkaSourceSplit;
+import org.apache.kafka.common.TopicPartition;
+
import javax.annotation.Nullable;
import java.util.ArrayList;
@@ -99,7 +102,8 @@ class ReaderRecoveryGate {
/**
* Sub-enumerators can assign newly discovered splits while reported
splits await reassignment.
* A checkpoint may therefore contain both pending reports and reader-held
splits, which must be
- * merged on restore. Preferring the current report for overlapping split
ids is defensive.
+ * merged on restore. Preferring the current report for the same physical
partition is
+ * defensive.
*/
private static List<DynamicKafkaSourceSplit> mergeReportedSplits(
@Nullable List<DynamicKafkaSourceSplit> previousReportedSplits,
@@ -107,14 +111,15 @@ class ReaderRecoveryGate {
if (previousReportedSplits == null ||
previousReportedSplits.isEmpty()) {
return new ArrayList<>(reportedSplits);
}
- Map<String, DynamicKafkaSourceSplit> mergedBySplitId = new
LinkedHashMap<>();
+ Map<Tuple2<String, TopicPartition>, DynamicKafkaSourceSplit>
mergedByPartition =
+ new LinkedHashMap<>();
for (DynamicKafkaSourceSplit split : previousReportedSplits) {
- mergedBySplitId.put(split.splitId(), split);
+
mergedByPartition.put(DynamicKafkaSourceEnumerator.toPartitionKey(split),
split);
}
for (DynamicKafkaSourceSplit split : reportedSplits) {
- mergedBySplitId.put(split.splitId(), split);
+
mergedByPartition.put(DynamicKafkaSourceEnumerator.toPartitionKey(split),
split);
}
- return new ArrayList<>(mergedBySplitId.values());
+ return new ArrayList<>(mergedByPartition.values());
}
/** Whether recovery gating is active and registrations must be deferred.
*/
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorRecoveryTest.java
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorRecoveryTest.java
index d20c2116..2379ba42 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorRecoveryTest.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorRecoveryTest.java
@@ -24,6 +24,7 @@ import
org.apache.flink.api.connector.source.SplitEnumeratorContext;
import org.apache.flink.api.connector.source.SplitsAssignment;
import
org.apache.flink.api.connector.source.SupportsSplitReassignmentOnRecovery;
import org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.dynamic.metadata.ClusterMetadata;
import org.apache.flink.connector.kafka.dynamic.metadata.KafkaMetadataService;
import org.apache.flink.connector.kafka.dynamic.metadata.KafkaStream;
@@ -66,6 +67,7 @@ import java.util.stream.Collectors;
import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Recovery tests for {@link DynamicKafkaSourceEnumerator}. */
public class DynamicKafkaSourceEnumeratorRecoveryTest {
@@ -990,6 +992,199 @@ public class DynamicKafkaSourceEnumeratorRecoveryTest {
Arguments.of(false, 1));
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void
testInitialRecoveryReRegistrationKeepsPartitionsWithCollidingSplitIds(
+ boolean bounded) throws Throwable {
+ DynamicKafkaSourceSplit activeSplit = createSplit("a", "b-c", 123L,
1000L);
+ DynamicKafkaSourceSplit retainedSplit =
+ createSplit("a-b", "c", 456L,
2000L).retainUntil(Long.MAX_VALUE);
+ assertThat(activeSplit.splitId()).isEqualTo(retainedSplit.splitId());
+ KafkaStream stream = createKafkaStream("stream", "a", "b-c");
+ List<DynamicKafkaSourceSplit> report = List.of(activeSplit,
retainedSplit);
+
+ try (MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
+ new MockSplitEnumeratorContext<>(2);
+ DynamicKafkaSourceEnumerator enumerator =
+ createEnumerator(
+ "stream",
+ new
MockKafkaMetadataService(Collections.singleton(stream)),
+ context,
+ createModeProperties("per_cluster"),
+ createRestoredState(stream, "a",
List.of(activeSplit)),
+ bounded)) {
+ enumerator.start();
+ registerReaderAndReportRestoredSplits(context, enumerator, 0,
report);
+ // Reader 0 fails while the initial gate still waits for reader 1.
+ context.unregisterReader(0);
+ enumerator.addSplitsBack(Collections.emptyList(), 0);
+ registerReaderAndReportRestoredSplits(context, enumerator, 0,
report);
+ registerReaderAndReportRestoredSplits(context, enumerator, 1,
Collections.emptyList());
+ context.runNextOneTimeCallable();
+ completePartitionDiscovery(getClusterEnumerator(enumerator, "a"));
+
+ assertExactCheckpointOffsets(context, 0, report);
+ assertSinglePhysicalOwners(context, 0, report.size());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void
testRestoredPendingReportsKeepPartitionsWithCollidingSplitIds(boolean bounded)
+ throws Throwable {
+ DynamicKafkaSourceSplit activeSplit = createSplit("a", "b-c", 123L,
1000L);
+ DynamicKafkaSourceSplit retainedSplit =
+ createSplit("a-b", "c", 456L,
2000L).retainUntil(Long.MAX_VALUE);
+ assertThat(activeSplit.splitId()).isEqualTo(retainedSplit.splitId());
+ KafkaStream stream = createKafkaStream("stream", "a", "b-c");
+ List<DynamicKafkaSourceSplit> report = List.of(activeSplit,
retainedSplit);
+ // A checkpoint taken while the initial gate was pending carries
reader 0's report.
+ DynamicKafkaSourceEnumState restoredState =
+ serdeRoundTrip(
+ new DynamicKafkaSourceEnumState(
+ Collections.singleton(stream),
+ Collections.singletonMap(
+ "a",
+ new KafkaSourceEnumState(
+
unwrapSplits(List.of(activeSplit)),
+ Collections.emptyList(),
+ true)),
+ Collections.emptyMap(),
+ Collections.singletonMap(0, report)));
+
+ try (MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
+ new MockSplitEnumeratorContext<>(2);
+ DynamicKafkaSourceEnumerator enumerator =
+ createEnumerator(
+ "stream",
+ new
MockKafkaMetadataService(Collections.singleton(stream)),
+ context,
+ createModeProperties("per_cluster"),
+ restoredState,
+ bounded)) {
+ enumerator.start();
+ registerReaderAndReportRestoredSplits(context, enumerator, 0,
report);
+ registerReaderAndReportRestoredSplits(context, enumerator, 1,
Collections.emptyList());
+ context.runNextOneTimeCallable();
+ completePartitionDiscovery(getClusterEnumerator(enumerator, "a"));
+
+ assertExactCheckpointOffsets(context, 0, report);
+ assertSinglePhysicalOwners(context, 0, report.size());
+ }
+ }
+
+ @ParameterizedTest
+ @MethodSource("collidingSplitModes")
+ public void testInitialRecoveryKeepsActivePartitionsWithCollidingSplitIds(
+ String mode, boolean bounded) throws Throwable {
+ DynamicKafkaSourceSplit firstSplit = createSplit("a", "b-c", 123L,
1000L);
+ DynamicKafkaSourceSplit secondSplit = createSplit("a-b", "c", 456L,
2000L);
+ assertThat(firstSplit.splitId()).isEqualTo(secondSplit.splitId());
+ KafkaStream stream = createKafkaStream("stream", Map.of("a", "b-c",
"a-b", "c"));
+ DynamicKafkaSourceEnumState restoredState =
+ createRestoredState(
+ stream, Map.of("a", List.of(firstSplit), "a-b",
List.of(secondSplit)));
+
+ try (MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
+ new MockSplitEnumeratorContext<>(2);
+ DynamicKafkaSourceEnumerator enumerator =
+ createEnumerator(
+ "stream",
+ new
MockKafkaMetadataService(Collections.singleton(stream)),
+ context,
+ createModeProperties(mode),
+ restoredState,
+ bounded)) {
+ enumerator.start();
+ registerReaderAndReportRestoredSplits(context, enumerator, 0,
List.of(firstSplit));
+ registerReaderAndReportRestoredSplits(context, enumerator, 1,
List.of(secondSplit));
+ context.runNextOneTimeCallable();
+ completePartitionDiscovery(getClusterEnumerator(enumerator, "a"));
+ completePartitionDiscovery(getClusterEnumerator(enumerator,
"a-b"));
+
+ assertExactCheckpointOffsets(context, 0, List.of(firstSplit,
secondSplit));
+ assertSinglePhysicalOwners(context, 0, 2);
+ }
+ }
+
+ @Test
+ public void testInitialRecoveryRejectsPartitionReportedByTwoReaders()
throws Throwable {
+ DynamicKafkaSourceSplit split = createSplit("cluster-0", "topic",
123L, 1000L);
+ KafkaStream stream = createKafkaStream("stream", "cluster-0", "topic");
+
+ try (MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
+ new MockSplitEnumeratorContext<>(2);
+ DynamicKafkaSourceEnumerator enumerator =
+ createEnumerator(
+ "stream",
+ new
MockKafkaMetadataService(Collections.singleton(stream)),
+ context,
+ createGlobalModeProperties(),
+ createRestoredState(stream, "cluster-0",
List.of(split)))) {
+ enumerator.start();
+ context.runNextOneTimeCallable();
+ registerReaderAndReportRestoredSplits(context, enumerator, 0,
List.of(split));
+
+ assertThatThrownBy(
+ () ->
+ registerReaderAndReportRestoredSplits(
+ context,
+ enumerator,
+ 1,
+ List.of(withOffset(split, 456L))))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("kafkaClusterId='cluster-0'")
+ .hasMessageContaining("reported by both reader 0 and
reader 1");
+ }
+ }
+
+ @ParameterizedTest
+ @MethodSource("collidingSplitModes")
+ public void testLocalRestartKeepsActivePartitionsWithCollidingSplitIds(
+ String mode, boolean bounded) throws Throwable {
+ DynamicKafkaSourceSplit firstSplit = createSplit("a", "b-c", 10L,
1000L);
+ DynamicKafkaSourceSplit secondSplit = createSplit("a-b", "c", 20L,
2000L);
+ assertThat(firstSplit.splitId()).isEqualTo(secondSplit.splitId());
+ KafkaStream stream = createKafkaStream("stream", Map.of("a", "b-c",
"a-b", "c"));
+
+ try (MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
+ new MockSplitEnumeratorContext<>(1);
+ DynamicKafkaSourceEnumerator enumerator =
+ createEnumerator(
+ "stream",
+ new
MockKafkaMetadataService(Collections.singleton(stream)),
+ context,
+ createModeProperties(mode),
+ new DynamicKafkaSourceEnumState(),
+ bounded)) {
+ enumerator.start();
+ registerReaderAndReportRestoredSplits(context, enumerator, 0,
Collections.emptyList());
+ context.runNextOneTimeCallable();
+ completePartitionDiscovery(
+ getClusterEnumerator(enumerator, "a"),
firstSplit.getKafkaPartitionSplit());
+ completePartitionDiscovery(
+ getClusterEnumerator(enumerator, "a-b"),
secondSplit.getKafkaPartitionSplit());
+ assertExactCheckpointOffsets(context, 0, List.of(firstSplit,
secondSplit));
+
+ int priorAssignments =
context.getSplitsAssignmentSequence().size();
+ List<DynamicKafkaSourceSplit> report =
+ List.of(withOffset(firstSplit, 500L),
withOffset(secondSplit, 600L));
+ context.unregisterReader(0);
+ enumerator.addSplitsBack(Collections.emptyList(), 0);
+ registerReaderAndReportRestoredSplits(context, enumerator, 0,
report);
+
+ assertExactCheckpointOffsets(context, priorAssignments, report);
+ }
+ }
+
+ private static Stream<Arguments> collidingSplitModes() {
+ return Stream.of(
+ Arguments.of("per_cluster", true),
+ Arguments.of("per_cluster", false),
+ Arguments.of("global", true),
+ Arguments.of("global", false));
+ }
+
@Test
public void testLocalFailurePreservesStillInactiveRetainedSplit() throws
Throwable {
String cluster = "cluster-0";
@@ -1645,7 +1840,15 @@ public class DynamicKafkaSourceEnumeratorRecoveryTest {
split.getKafkaPartitionSplit().getTopicPartition(),
offset));
}
- private static void completePartitionDiscovery(KafkaSourceEnumerator
enumerator)
+ private static KafkaSourceEnumerator getClusterEnumerator(
+ DynamicKafkaSourceEnumerator enumerator, String clusterId) {
+ return (KafkaSourceEnumerator)
+ ((Map<?, ?>) Whitebox.getInternalState(enumerator,
"clusterEnumeratorMap"))
+ .get(clusterId);
+ }
+
+ private static void completePartitionDiscovery(
+ KafkaSourceEnumerator enumerator, KafkaPartitionSplit...
discoveredSplits)
throws Exception {
Class<?> changeClass =
Class.forName(KafkaSourceEnumerator.class.getName() +
"$PartitionSplitChange");
@@ -1657,7 +1860,7 @@ public class DynamicKafkaSourceEnumeratorRecoveryTest {
handler.setAccessible(true);
handler.invoke(
enumerator,
- constructor.newInstance(Collections.emptySet(),
Collections.emptySet()),
+ constructor.newInstance(Set.of(discoveredSplits),
Collections.emptySet()),
null);
}
@@ -1678,7 +1881,7 @@ public class DynamicKafkaSourceEnumeratorRecoveryTest {
MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context,
int skipAssignments,
int expectedPartitions) {
- Map<String, Set<Integer>> owners = new HashMap<>();
+ Map<Tuple2<String, TopicPartition>, Set<Integer>> owners = new
HashMap<>();
context.getSplitsAssignmentSequence().stream()
.skip(skipAssignments)
.forEach(
@@ -1690,8 +1893,9 @@ public class DynamicKafkaSourceEnumeratorRecoveryTest {
splits.forEach(
split ->
owners.computeIfAbsent(
-
split
-
.splitId(),
+
DynamicKafkaSourceEnumerator
+
.toPartitionKey(
+
split),
ignored ->
new HashSet<>())
.add(reader))));
@@ -1762,19 +1966,31 @@ public class DynamicKafkaSourceEnumeratorRecoveryTest {
private static DynamicKafkaSourceEnumState createRestoredState(
KafkaStream kafkaStream, String clusterId,
List<DynamicKafkaSourceSplit> activeSplits) {
- return new DynamicKafkaSourceEnumState(
- Collections.singleton(kafkaStream),
- Collections.singletonMap(
- clusterId,
- new KafkaSourceEnumState(
- unwrapSplits(activeSplits),
Collections.emptyList(), true)));
+ return createRestoredState(kafkaStream,
Collections.singletonMap(clusterId, activeSplits));
+ }
+
+ private static DynamicKafkaSourceEnumState createRestoredState(
+ KafkaStream kafkaStream, Map<String,
List<DynamicKafkaSourceSplit>> splitsByCluster) {
+ Map<String, KafkaSourceEnumState> clusterStates = new HashMap<>();
+ splitsByCluster.forEach(
+ (clusterId, activeSplits) ->
+ clusterStates.put(
+ clusterId,
+ new KafkaSourceEnumState(
+ unwrapSplits(activeSplits),
+ Collections.emptyList(),
+ true)));
+ return new
DynamicKafkaSourceEnumState(Collections.singleton(kafkaStream), clusterStates);
}
private static Properties createGlobalModeProperties() {
- Properties properties = createBaseProperties();
- properties.setProperty(
- DynamicKafkaSourceOptions.STREAM_ENUMERATOR_MODE.key(),
+ return createModeProperties(
DynamicKafkaSourceOptions.EnumeratorMode.GLOBAL.name().toLowerCase());
+ }
+
+ private static Properties createModeProperties(String mode) {
+ Properties properties = createBaseProperties();
+
properties.setProperty(DynamicKafkaSourceOptions.STREAM_ENUMERATOR_MODE.key(),
mode);
return properties;
}
@@ -1792,14 +2008,25 @@ public class DynamicKafkaSourceEnumeratorRecoveryTest {
SplitEnumeratorContext<DynamicKafkaSourceSplit> context,
Properties properties,
DynamicKafkaSourceEnumState restoredState) {
+ return createEnumerator(
+ streamId, metadataService, context, properties, restoredState,
false);
+ }
+
+ private static DynamicKafkaSourceEnumerator createEnumerator(
+ String streamId,
+ KafkaMetadataService metadataService,
+ SplitEnumeratorContext<DynamicKafkaSourceSplit> context,
+ Properties properties,
+ DynamicKafkaSourceEnumState restoredState,
+ boolean bounded) {
return new DynamicKafkaSourceEnumerator(
new KafkaStreamSetSubscriber(Collections.singleton(streamId)),
metadataService,
context,
OffsetsInitializer.earliest(),
- new NoStoppingOffsetsInitializer(),
+ bounded ? OffsetsInitializer.latest() : new
NoStoppingOffsetsInitializer(),
properties,
- Boundedness.CONTINUOUS_UNBOUNDED,
+ bounded ? Boundedness.BOUNDED :
Boundedness.CONTINUOUS_UNBOUNDED,
restoredState,
new NoOpKafkaEnumContextProxyFactory());
}
@@ -1809,6 +2036,21 @@ public class DynamicKafkaSourceEnumeratorRecoveryTest {
return createKafkaStream(streamId, clusterId,
Collections.singleton(activeTopic));
}
+ private static KafkaStream createKafkaStream(
+ String streamId, Map<String, String> activeTopicByCluster) {
+ Properties clusterProperties = new Properties();
+ clusterProperties.setProperty(
+ CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
"localhost:9092");
+ Map<String, ClusterMetadata> clusterMetadata = new HashMap<>();
+ activeTopicByCluster.forEach(
+ (clusterId, topic) ->
+ clusterMetadata.put(
+ clusterId,
+ new ClusterMetadata(
+ Collections.singleton(topic),
clusterProperties)));
+ return new KafkaStream(streamId, clusterMetadata);
+ }
+
private static KafkaStream createKafkaStream(
String streamId, String clusterId, Set<String> activeTopics) {
Properties clusterProperties = new Properties();
@@ -1820,6 +2062,14 @@ public class DynamicKafkaSourceEnumeratorRecoveryTest {
clusterId, new ClusterMetadata(activeTopics,
clusterProperties)));
}
+ private static DynamicKafkaSourceSplit createSplit(
+ String clusterId, String topic, long startingOffset, long
stoppingOffset) {
+ return new DynamicKafkaSourceSplit(
+ clusterId,
+ new KafkaPartitionSplit(
+ new TopicPartition(topic, 0), startingOffset,
stoppingOffset));
+ }
+
private static List<DynamicKafkaSourceSplit> createSplits(
String clusterId, String topic, int count) {
List<DynamicKafkaSourceSplit> splits = new ArrayList<>();
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/ReaderRecoveryGateTest.java
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/ReaderRecoveryGateTest.java
index 4b2cfa51..fce2f6d7 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/ReaderRecoveryGateTest.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/ReaderRecoveryGateTest.java
@@ -200,6 +200,29 @@ class ReaderRecoveryGateTest {
assertThat(gate.drainReportedSplits(1)).isEmpty();
}
+ @Test
+ void testMergeKeepsDistinctPartitionsWithCollidingSplitIds() {
+ // Cluster "a" topic "b-c" and cluster "a-b" topic "c" both format to
split id "a-b-c-0".
+ DynamicKafkaSourceSplit firstSplit =
+ new DynamicKafkaSourceSplit(
+ "a", new KafkaPartitionSplit(new TopicPartition("b-c",
0), 123L));
+ DynamicKafkaSourceSplit secondSplit =
+ new DynamicKafkaSourceSplit(
+ "a-b", new KafkaPartitionSplit(new TopicPartition("c",
0), 456L));
+ DynamicKafkaSourceSplit returnedSplit =
+ new DynamicKafkaSourceSplit(
+ "a-b", new KafkaPartitionSplit(new TopicPartition("c",
0), 789L));
+ assertThat(firstSplit.splitId()).isEqualTo(secondSplit.splitId());
+
+ ReaderRecoveryGate gate = new ReaderRecoveryGate(true);
+ gate.recordReportedSplits(0, Collections.singletonList(firstSplit));
+ gate.recordReportedSplits(0, Collections.singletonList(secondSplit));
+
assertThat(gate.snapshotReportedSplits().get(0)).containsExactly(firstSplit,
secondSplit);
+
+ gate.recordReturnedSplits(0, Collections.singletonList(returnedSplit));
+
assertThat(gate.snapshotReportedSplits().get(0)).containsExactly(firstSplit,
returnedSplit);
+ }
+
@Test
void testDrainReportedSplitsReturnsReaderOrderAndClears() {
ReaderRecoveryGate gate = new ReaderRecoveryGate(true);
