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 e79fe983 [FLINK-40469] Checkpoint during recovery window silently
loses partitions (#295)
e79fe983 is described below
commit e79fe9837f15c672172d43848e3d328ff96fe51a
Author: Aleksandr Savonin <[email protected]>
AuthorDate: Sun Sep 20 15:27:08 2026 +0200
[FLINK-40469] Checkpoint during recovery window silently loses partitions
(#295)
[FLINK-40469][Connectors/Kafka] Checkpoint reported splits pending recovery
reassignment
Under SupportsSplitReassignmentOnRecovery, restored readers hold no
splits and report them to DynamicKafkaSourceEnumerator, which defers
reassignment until the first metadata discovery completes and all
readers register. A checkpoint completing inside that window captured
no record of the reported splits: the readers' state was empty and the
sub enumerator states still marked the partitions assigned. Restoring
from such a checkpoint silently stopped consuming those partitions.
Checkpoint the pending reports in DynamicKafkaSourceEnumState and
restore them in the enumerator constructor, remapping reader ids onto
the current parallelism and merging a reader's registration report with
any restored entry per split id. The merge is needed because a restored
sub enumerator can assign a newly discovered partition to a reader whose
earlier report is still pending, so one checkpoint can hold both.
DynamicKafkaSourceEnumStateSerializer goes to VERSION_4, append-only:
V1 to V3 checkpoints restore unchanged with an empty pending map.
Checkpoints written by V4 cannot be read by older connector versions.
Generated-by: Codex CLI 0.155.0-alpha.2.6
Generated-by: Codex CLI 0.155.0-alpha.9
---
AGENTS.md | 2 +-
.../enumerator/DynamicKafkaSourceEnumState.java | 30 +-
.../DynamicKafkaSourceEnumStateSerializer.java | 63 ++-
.../enumerator/DynamicKafkaSourceEnumerator.java | 45 ++-
.../source/enumerator/ReaderRecoveryGate.java | 39 +-
.../DynamicKafkaSourceEnumStateSerializerTest.java | 98 ++++-
.../DynamicKafkaSourceEnumeratorRecoveryTest.java | 435 +++++++++++++++++++++
.../source/enumerator/ReaderRecoveryGateTest.java | 38 +-
.../DynamicKafkaSourceEnumStateTestUtils.java | 64 +++
9 files changed, 793 insertions(+), 21 deletions(-)
diff --git a/AGENTS.md b/AGENTS.md
index c048ee5c..47ed9b5d 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -122,7 +122,7 @@ This file provides guidance for AI coding agents working
with the Apache Flink K
### Changing checkpointed state
-Splits, enumerator state, writer state and committables are written by
`SimpleVersionedSerializer` implementations. Current versions:
`KafkaPartitionSplitSerializer` 0, `KafkaSourceEnumStateSerializer` 4,
`KafkaWriterStateSerializer` 2, `KafkaCommittableSerializer` 1,
`DynamicKafkaSourceSplitSerializer` 2, `DynamicKafkaSourceEnumStateSerializer`
3.
+Splits, enumerator state, writer state and committables are written by
`SimpleVersionedSerializer` implementations. Current versions:
`KafkaPartitionSplitSerializer` 0, `KafkaSourceEnumStateSerializer` 4,
`KafkaWriterStateSerializer` 2, `KafkaCommittableSerializer` 1,
`DynamicKafkaSourceSplitSerializer` 2, `DynamicKafkaSourceEnumStateSerializer`
4.
1. Bump the version and keep a read path for every older version
2. Enum ordinals written to state (for example `TransactionOwnership`) must
not be reordered
diff --git
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumState.java
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumState.java
index bb474d8c..d5eab749 100644
---
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumState.java
+++
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumState.java
@@ -20,10 +20,12 @@ package
org.apache.flink.connector.kafka.dynamic.source.enumerator;
import org.apache.flink.annotation.Internal;
import org.apache.flink.connector.kafka.dynamic.metadata.KafkaStream;
+import
org.apache.flink.connector.kafka.dynamic.source.split.DynamicKafkaSourceSplit;
import org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumState;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -36,26 +38,27 @@ public class DynamicKafkaSourceEnumState {
private final Set<KafkaStream> kafkaStreams;
private final Map<String, KafkaSourceEnumState> clusterEnumeratorStates;
private final Map<String, RetainedClusterState>
retainedClusterEnumeratorStates;
+ private final Map<Integer, List<DynamicKafkaSourceSplit>>
pendingReportedSplitsByReader;
public DynamicKafkaSourceEnumState() {
- this.kafkaStreams = new HashSet<>();
- this.clusterEnumeratorStates = new HashMap<>();
- this.retainedClusterEnumeratorStates = new HashMap<>();
+ this(new HashSet<>(), new HashMap<>(), new HashMap<>(), new
HashMap<>());
}
public DynamicKafkaSourceEnumState(
Set<KafkaStream> kafkaStreams,
Map<String, KafkaSourceEnumState> clusterEnumeratorStates) {
- this(kafkaStreams, clusterEnumeratorStates, new HashMap<>());
+ this(kafkaStreams, clusterEnumeratorStates, new HashMap<>(), new
HashMap<>());
}
public DynamicKafkaSourceEnumState(
Set<KafkaStream> kafkaStreams,
Map<String, KafkaSourceEnumState> clusterEnumeratorStates,
- Map<String, RetainedClusterState> retainedClusterEnumeratorStates)
{
+ Map<String, RetainedClusterState> retainedClusterEnumeratorStates,
+ Map<Integer, List<DynamicKafkaSourceSplit>>
pendingReportedSplitsByReader) {
this.kafkaStreams = kafkaStreams;
this.clusterEnumeratorStates = clusterEnumeratorStates;
this.retainedClusterEnumeratorStates = retainedClusterEnumeratorStates;
+ this.pendingReportedSplitsByReader = pendingReportedSplitsByReader;
}
public Set<KafkaStream> getKafkaStreams() {
@@ -70,6 +73,23 @@ public class DynamicKafkaSourceEnumState {
return retainedClusterEnumeratorStates;
}
+ /**
+ * Splits reported by readers on registration that were not yet reassigned
when the checkpoint
+ * was taken. Readers hold no splits at that point and the sub enumerator
states already mark
+ * the partitions as assigned, so until reassignment this map is the only
record of the
+ * reader-reported offsets.
+ *
+ * <p>Only reports the enumerator has already received need to be
checkpointed. Per the FLIP-537
+ * contract, the coordinator receives a restarting reader's report before
any subsequent
+ * checkpoint request: the report is sent before the reader's task turns
{@code RUNNING}, and no
+ * checkpoint can be triggered until it is. A checkpoint that overlaps a
reader failure can only
+ * complete if the source task acknowledged it before failing, making it a
consistent
+ * pre-failure snapshot.
+ */
+ public Map<Integer, List<DynamicKafkaSourceSplit>>
getPendingReportedSplitsByReader() {
+ return pendingReportedSplitsByReader;
+ }
+
/** Kafka enumerator state that stays checkpointed after its cluster
becomes inactive. */
public static class RetainedClusterState {
private final KafkaSourceEnumState kafkaSourceEnumState;
diff --git
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumStateSerializer.java
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumStateSerializer.java
index 1cec2a18..269da01a 100644
---
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumStateSerializer.java
+++
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumStateSerializer.java
@@ -21,6 +21,8 @@ package
org.apache.flink.connector.kafka.dynamic.source.enumerator;
import org.apache.flink.annotation.Internal;
import org.apache.flink.connector.kafka.dynamic.metadata.ClusterMetadata;
import org.apache.flink.connector.kafka.dynamic.metadata.KafkaStream;
+import
org.apache.flink.connector.kafka.dynamic.source.split.DynamicKafkaSourceSplit;
+import
org.apache.flink.connector.kafka.dynamic.source.split.DynamicKafkaSourceSplitSerializer;
import org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumState;
import
org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumStateSerializer;
import
org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
@@ -37,8 +39,10 @@ import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
@@ -51,16 +55,19 @@ public class DynamicKafkaSourceEnumStateSerializer
private static final int VERSION_1 = 1;
private static final int VERSION_2 = 2;
private static final int VERSION_3 = 3;
+ private static final int VERSION_4 = 4;
private final KafkaSourceEnumStateSerializer
kafkaSourceEnumStateSerializer;
+ private final DynamicKafkaSourceSplitSerializer splitSerializer;
public DynamicKafkaSourceEnumStateSerializer() {
this.kafkaSourceEnumStateSerializer = new
KafkaSourceEnumStateSerializer();
+ this.splitSerializer = new DynamicKafkaSourceSplitSerializer();
}
@Override
public int getVersion() {
- return VERSION_3;
+ return VERSION_4;
}
@Override
@@ -88,6 +95,8 @@ public class DynamicKafkaSourceEnumStateSerializer
retainedClusterEnumeratorState.getValue().getKafkaSourceEnumState(), out);
}
+
writePendingReportedSplits(state.getPendingReportedSplitsByReader(), out);
+
return baos.toByteArray();
}
}
@@ -95,7 +104,7 @@ public class DynamicKafkaSourceEnumStateSerializer
@Override
public DynamicKafkaSourceEnumState deserialize(int version, byte[]
serialized)
throws IOException {
- if (version != VERSION_1 && version != VERSION_2 && version !=
VERSION_3) {
+ if (version < VERSION_1 || version > getVersion()) {
throw new IOException(
String.format(
"The bytes are serialized with version %d, "
@@ -116,7 +125,7 @@ public class DynamicKafkaSourceEnumStateSerializer
Map<String, DynamicKafkaSourceEnumState.RetainedClusterState>
retainedClusterEnumeratorStates = new HashMap<>();
- if (version == VERSION_3) {
+ if (version >= VERSION_3) {
int retainedClusterEnumeratorStateMapSize = in.readInt();
for (int i = 0; i < retainedClusterEnumeratorStateMapSize;
i++) {
String kafkaClusterId = in.readUTF();
@@ -130,9 +139,55 @@ public class DynamicKafkaSourceEnumStateSerializer
}
}
+ Map<Integer, List<DynamicKafkaSourceSplit>>
pendingReportedSplitsByReader =
+ version >= VERSION_4 ? readPendingReportedSplits(in) : new
HashMap<>();
+
return new DynamicKafkaSourceEnumState(
- kafkaStreams, clusterEnumeratorStates,
retainedClusterEnumeratorStates);
+ kafkaStreams,
+ clusterEnumeratorStates,
+ retainedClusterEnumeratorStates,
+ pendingReportedSplitsByReader);
+ }
+ }
+
+ private void writePendingReportedSplits(
+ Map<Integer, List<DynamicKafkaSourceSplit>>
pendingReportedSplitsByReader,
+ DataOutputStream out)
+ throws IOException {
+ out.writeInt(splitSerializer.getVersion());
+ out.writeInt(pendingReportedSplitsByReader.size());
+ for (Map.Entry<Integer, List<DynamicKafkaSourceSplit>> readerSplits :
+ pendingReportedSplitsByReader.entrySet()) {
+ out.writeInt(readerSplits.getKey());
+ out.writeInt(readerSplits.getValue().size());
+ for (DynamicKafkaSourceSplit split : readerSplits.getValue()) {
+ byte[] splitBytes = splitSerializer.serialize(split);
+ // DynamicKafkaSourceSplitSerializer consumes all remaining
bytes on
+ // deserialization, so each split must be length-prefixed.
+ out.writeInt(splitBytes.length);
+ out.write(splitBytes);
+ }
+ }
+ }
+
+ private Map<Integer, List<DynamicKafkaSourceSplit>>
readPendingReportedSplits(
+ DataInputStream in) throws IOException {
+ Map<Integer, List<DynamicKafkaSourceSplit>>
pendingReportedSplitsByReader = new HashMap<>();
+ int splitSerializerVersion = in.readInt();
+ int readerCount = in.readInt();
+ for (int i = 0; i < readerCount; i++) {
+ int readerId = in.readInt();
+ int splitCount = in.readInt();
+ List<DynamicKafkaSourceSplit> splits = new ArrayList<>(splitCount);
+ for (int j = 0; j < splitCount; j++) {
+ int splitByteSize = in.readInt();
+ splits.add(
+ splitSerializer.deserialize(
+ splitSerializerVersion, readNBytes(in,
splitByteSize)));
+ }
+ pendingReportedSplitsByReader.put(readerId, splits);
}
+ return pendingReportedSplitsByReader;
}
private void writeClusterEnumeratorStates(
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 ea38c9e6..6cc98d91 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
@@ -218,6 +218,8 @@ public class DynamicKafkaSourceEnumerator
this.splitAssignmentStrategy =
createSplitAssignmentStrategy(properties);
this.readerRecoveryGate =
new
ReaderRecoveryGate(hasRestoredEnumeratorState(dynamicKafkaSourceEnumState));
+
restorePendingReportedSplits(dynamicKafkaSourceEnumState.getPendingReportedSplitsByReader())
+ .forEach(readerRecoveryGate::recordReportedSplits);
if
(!dynamicKafkaSourceEnumState.getClusterEnumeratorStates().isEmpty()) {
logger.info("Dynamic Kafka source restored from checkpointed
enumerator state");
@@ -291,6 +293,32 @@ public class DynamicKafkaSourceEnumerator
splitAssignmentStrategy.onMetadataRefresh(activeSplitIds);
}
+ /**
+ * Restores reported splits that a checkpoint captured before recovery
reassignment ran. Reader
+ * ids from a checkpoint taken at a different parallelism are remapped
onto the current one, and
+ * entries that collapse onto the same reader are merged.
+ */
+ private Map<Integer, List<DynamicKafkaSourceSplit>>
restorePendingReportedSplits(
+ Map<Integer, List<DynamicKafkaSourceSplit>>
restoredPendingReportedSplits) {
+ Map<Integer, List<DynamicKafkaSourceSplit>> remappedSplitsByReader =
new HashMap<>();
+ if (restoredPendingReportedSplits.isEmpty()) {
+ return remappedSplitsByReader;
+ }
+ int parallelism = enumContext.currentParallelism();
+ for (Entry<Integer, List<DynamicKafkaSourceSplit>> readerSplits :
+ restoredPendingReportedSplits.entrySet()) {
+ int readerId = Math.floorMod(readerSplits.getKey(), parallelism);
+ remappedSplitsByReader
+ .computeIfAbsent(readerId, ignored -> new ArrayList<>())
+ .addAll(readerSplits.getValue());
+ }
+ logger.info(
+ "Restored {} reported splits that were pending reassignment
when the checkpoint"
+ + " was taken",
+
restoredPendingReportedSplits.values().stream().mapToInt(List::size).sum());
+ return remappedSplitsByReader;
+ }
+
private Set<KafkaStream>
refreshRestoredClusterPropertiesFromMetadataService(
Set<KafkaStream> restoredKafkaStreams) {
Set<KafkaStream> fetchedKafkaStreams =
@@ -923,7 +951,8 @@ public class DynamicKafkaSourceEnumerator
private static boolean hasRestoredEnumeratorState(
DynamicKafkaSourceEnumState dynamicKafkaSourceEnumState) {
return
!dynamicKafkaSourceEnumState.getClusterEnumeratorStates().isEmpty()
- ||
!dynamicKafkaSourceEnumState.getRetainedClusterEnumeratorStates().isEmpty();
+ ||
!dynamicKafkaSourceEnumState.getRetainedClusterEnumeratorStates().isEmpty()
+ ||
!dynamicKafkaSourceEnumState.getPendingReportedSplitsByReader().isEmpty();
}
private static class ReportedSplit {
@@ -967,10 +996,22 @@ public class DynamicKafkaSourceEnumerator
}
}
+ Map<Integer, List<DynamicKafkaSourceSplit>>
pendingReportedSplitsByReader =
+ readerRecoveryGate.snapshotReportedSplits();
+ if (isCheckpointSnapshot && !pendingReportedSplitsByReader.isEmpty()) {
+ logger.debug(
+ "Checkpoint {} includes pending reported splits of readers
{}",
+ checkpointId,
+ pendingReportedSplitsByReader.keySet());
+ }
+
+ // See DynamicKafkaSourceEnumState#getPendingReportedSplitsByReader()
for why the
+ // pending splits are checkpointed.
return new DynamicKafkaSourceEnumState(
latestKafkaStreams,
subEnumeratorStateByCluster,
- new HashMap<>(retainedClusterEnumeratorStates));
+ new HashMap<>(retainedClusterEnumeratorStates),
+ pendingReportedSplitsByReader);
}
private void retainRemovedClusterEnumeratorStates(
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 5707a078..14bc6930 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
@@ -21,11 +21,14 @@ package
org.apache.flink.connector.kafka.dynamic.source.enumerator;
import org.apache.flink.annotation.Internal;
import
org.apache.flink.connector.kafka.dynamic.source.split.DynamicKafkaSourceSplit;
+import javax.annotation.Nullable;
+
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
@@ -72,11 +75,43 @@ class ReaderRecoveryGate {
this.initialReaderRegistrationPending = restoredFromCheckpoint;
}
- /** Records splits a reader reported on registration; an empty report is
ignored. */
+ /** Merges a reader's registration report with pending splits; an empty
report is ignored. */
void recordReportedSplits(int subtaskId, List<DynamicKafkaSourceSplit>
reportedSplits) {
if (!reportedSplits.isEmpty()) {
- pendingReportedSplitsByReader.put(subtaskId, new
ArrayList<>(reportedSplits));
+ pendingReportedSplitsByReader.put(
+ subtaskId,
+ mergeReportedSplits(
+ pendingReportedSplitsByReader.get(subtaskId),
reportedSplits));
+ }
+ }
+
+ /** Returns a copy of the pending reports without draining them. */
+ Map<Integer, List<DynamicKafkaSourceSplit>> snapshotReportedSplits() {
+ Map<Integer, List<DynamicKafkaSourceSplit>> snapshot = new HashMap<>();
+ pendingReportedSplitsByReader.forEach(
+ (readerId, splits) -> snapshot.put(readerId, new
ArrayList<>(splits)));
+ return snapshot;
+ }
+
+ /**
+ * 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.
+ */
+ private static List<DynamicKafkaSourceSplit> mergeReportedSplits(
+ @Nullable List<DynamicKafkaSourceSplit> previousReportedSplits,
+ List<DynamicKafkaSourceSplit> reportedSplits) {
+ if (previousReportedSplits == null ||
previousReportedSplits.isEmpty()) {
+ return new ArrayList<>(reportedSplits);
+ }
+ Map<String, DynamicKafkaSourceSplit> mergedBySplitId = new
LinkedHashMap<>();
+ for (DynamicKafkaSourceSplit split : previousReportedSplits) {
+ mergedBySplitId.put(split.splitId(), split);
+ }
+ for (DynamicKafkaSourceSplit split : reportedSplits) {
+ mergedBySplitId.put(split.splitId(), split);
}
+ return new ArrayList<>(mergedBySplitId.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/DynamicKafkaSourceEnumStateSerializerTest.java
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumStateSerializerTest.java
index d79e9323..51ca3108 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumStateSerializerTest.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumStateSerializerTest.java
@@ -20,6 +20,7 @@ package
org.apache.flink.connector.kafka.dynamic.source.enumerator;
import org.apache.flink.connector.kafka.dynamic.metadata.ClusterMetadata;
import org.apache.flink.connector.kafka.dynamic.metadata.KafkaStream;
+import
org.apache.flink.connector.kafka.dynamic.source.split.DynamicKafkaSourceSplit;
import
org.apache.flink.connector.kafka.dynamic.source.testutils.DynamicKafkaSourceEnumStateTestUtils;
import org.apache.flink.connector.kafka.source.enumerator.AssignmentStatus;
import org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumState;
@@ -27,6 +28,7 @@ import
org.apache.flink.connector.kafka.source.enumerator.SplitAndAssignmentStat
import
org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.connector.kafka.source.split.KafkaPartitionSplit;
+import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import org.apache.kafka.clients.CommonClientConfigs;
@@ -112,7 +114,25 @@ public class DynamicKafkaSourceEnumStateSerializerTest {
ImmutableSet.of(
getSplitAssignment("topic6", 0, ASSIGNED)),
true),
- 123L)));
+ 123L)),
+ ImmutableMap.of(
+ 0,
+ ImmutableList.of(
+ new DynamicKafkaSourceSplit(
+ "cluster0",
+ new KafkaPartitionSplit(
+ new
TopicPartition("topic0", 0), 100))),
+ 2,
+ ImmutableList.of(
+ new DynamicKafkaSourceSplit(
+ "cluster1",
+ new KafkaPartitionSplit(
+ new
TopicPartition("topic2", 0), 5)),
+ new DynamicKafkaSourceSplit(
+ "retained-cluster",
+ new KafkaPartitionSplit(
+ new
TopicPartition("topic6", 0), 7),
+ 456L))));
DynamicKafkaSourceEnumState dynamicKafkaSourceEnumStateAfterSerde =
dynamicKafkaSourceEnumStateSerializer.deserialize(
@@ -212,6 +232,82 @@ public class DynamicKafkaSourceEnumStateSerializerTest {
assertThat(clusterMetadata.getStoppingOffsetsInitializer()).isNull();
}
+ @Test
+ public void testDeserializeV2State() throws Exception {
+ DynamicKafkaSourceEnumStateSerializer
dynamicKafkaSourceEnumStateSerializer =
+ new DynamicKafkaSourceEnumStateSerializer();
+ KafkaSourceEnumState clusterState =
+ new KafkaSourceEnumState(
+ ImmutableSet.of(
+ getSplitAssignment("topic0", 0, ASSIGNED),
+ getSplitAssignment("topic0", 1, UNASSIGNED)),
+ true);
+
+ byte[] serializedState =
+ DynamicKafkaSourceEnumStateTestUtils.serializeV2State(
+ "stream0",
+ "cluster0",
+ ImmutableSet.of("topic0"),
+ "cluster0:9092",
+ clusterState);
+
+ DynamicKafkaSourceEnumState dynamicKafkaSourceEnumState =
+ dynamicKafkaSourceEnumStateSerializer.deserialize(2,
serializedState);
+
+ assertThat(dynamicKafkaSourceEnumState.getClusterEnumeratorStates())
+ .containsOnlyKeys("cluster0");
+
assertThat(dynamicKafkaSourceEnumState.getClusterEnumeratorStates().get("cluster0"))
+ .usingRecursiveComparison()
+ .isEqualTo(clusterState);
+ KafkaStream kafkaStream =
dynamicKafkaSourceEnumState.getKafkaStreams().iterator().next();
+ assertThat(kafkaStream.getStreamId()).isEqualTo("stream0");
+
assertThat(dynamicKafkaSourceEnumState.getRetainedClusterEnumeratorStates()).isEmpty();
+
assertThat(dynamicKafkaSourceEnumState.getPendingReportedSplitsByReader()).isEmpty();
+ }
+
+ @Test
+ public void testDeserializeV3State() throws Exception {
+ DynamicKafkaSourceEnumStateSerializer
dynamicKafkaSourceEnumStateSerializer =
+ new DynamicKafkaSourceEnumStateSerializer();
+ KafkaSourceEnumState clusterState =
+ new KafkaSourceEnumState(
+ ImmutableSet.of(getSplitAssignment("topic0", 0,
ASSIGNED)), true);
+ KafkaSourceEnumState retainedClusterState =
+ new KafkaSourceEnumState(
+ ImmutableSet.of(getSplitAssignment("topic6", 0,
ASSIGNED)), true);
+
+ byte[] serializedState =
+ DynamicKafkaSourceEnumStateTestUtils.serializeV3State(
+ "stream0",
+ "cluster0",
+ ImmutableSet.of("topic0"),
+ "cluster0:9092",
+ clusterState,
+ "retained-cluster",
+ 123L,
+ retainedClusterState);
+
+ DynamicKafkaSourceEnumState dynamicKafkaSourceEnumState =
+ dynamicKafkaSourceEnumStateSerializer.deserialize(3,
serializedState);
+
+ assertThat(dynamicKafkaSourceEnumState.getClusterEnumeratorStates())
+ .containsOnlyKeys("cluster0");
+
assertThat(dynamicKafkaSourceEnumState.getClusterEnumeratorStates().get("cluster0"))
+ .usingRecursiveComparison()
+ .isEqualTo(clusterState);
+
assertThat(dynamicKafkaSourceEnumState.getRetainedClusterEnumeratorStates())
+ .containsOnlyKeys("retained-cluster");
+ DynamicKafkaSourceEnumState.RetainedClusterState retainedState =
+ dynamicKafkaSourceEnumState
+ .getRetainedClusterEnumeratorStates()
+ .get("retained-cluster");
+ assertThat(retainedState.getRetainedUntilMs()).isEqualTo(123L);
+ assertThat(retainedState.getKafkaSourceEnumState())
+ .usingRecursiveComparison()
+ .isEqualTo(retainedClusterState);
+
assertThat(dynamicKafkaSourceEnumState.getPendingReportedSplitsByReader()).isEmpty();
+ }
+
private static SplitAndAssignmentStatus getSplitAssignment(
String topic, int partition, AssignmentStatus assignStatus) {
return new SplitAndAssignmentStatus(
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 9eb84b3a..88adb63f 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
@@ -238,6 +238,441 @@ public class DynamicKafkaSourceEnumeratorRecoveryTest {
.isTrue();
}
+ @Test
+ public void
testCheckpointDuringPendingRecoveryPreservesSplitsAndReportedOffsets()
+ throws Throwable {
+ int parallelism = 2;
+ String streamId = "stream";
+ String clusterId = "cluster-0";
+ String topic = "topic";
+ List<DynamicKafkaSourceSplit> enumeratorStateSplits =
createSplits(clusterId, topic, 2);
+ List<DynamicKafkaSourceSplit> reportedSplits = new ArrayList<>();
+ // use distinct offsets to verify that recovery preserves the
reader-reported positions
+ // instead of the stale EARLIEST offsets
+ for (int partition = 0; partition < 2; partition++) {
+ reportedSplits.add(
+ new DynamicKafkaSourceSplit(
+ clusterId,
+ new KafkaPartitionSplit(
+ new TopicPartition(topic, partition), 42 +
partition)));
+ }
+ KafkaStream kafkaStream = createKafkaStream(streamId, clusterId,
topic);
+ Properties properties = createGlobalModeProperties();
+
+ DynamicKafkaSourceEnumState inWindowCheckpoint;
+ try (MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
+ new MockSplitEnumeratorContext<>(parallelism);
+ DynamicKafkaSourceEnumerator enumerator =
+ createEnumerator(
+ streamId,
+ new
MockKafkaMetadataService(Collections.singleton(kafkaStream)),
+ context,
+ properties,
+ createRestoredState(
+ kafkaStream, clusterId,
enumeratorStateSplits))) {
+ enumerator.start();
+ context.registerReader(ReaderInfo.createReaderInfo(0,
"location-0", reportedSplits));
+ enumerator.addReader(0);
+ context.registerReader(
+ ReaderInfo.createReaderInfo(1, "location-1",
Collections.emptyList()));
+ enumerator.addReader(1);
+
+ assertThat(context.getSplitsAssignmentSequence()).isEmpty();
+ inWindowCheckpoint = enumerator.snapshotState(1L);
+ }
+
+ DynamicKafkaSourceEnumState restoredCheckpoint =
serdeRoundTrip(inWindowCheckpoint);
+ try (MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
+ new MockSplitEnumeratorContext<>(parallelism);
+ DynamicKafkaSourceEnumerator enumerator =
+ createEnumerator(
+ streamId,
+ new
MockKafkaMetadataService(Collections.singleton(kafkaStream)),
+ context,
+ properties,
+ restoredCheckpoint)) {
+ enumerator.start();
+ registerReadersWithoutReportedSplits(context, enumerator,
parallelism);
+ context.runNextOneTimeCallable();
+
+ assertAssignedSplitIdsExactly(context, reportedSplits);
+ Map<Integer, Long> startingOffsetByPartition = new HashMap<>();
+ for (DynamicKafkaSourceSplit split :
collectAssignedSplits(context)) {
+ startingOffsetByPartition.put(
+ split.getKafkaPartitionSplit().getPartition(),
+ split.getKafkaPartitionSplit().getStartingOffset());
+ }
+ assertThat(startingOffsetByPartition)
+ .as("assigned splits must carry the reader-reported
offsets")
+ .containsEntry(0, 42L)
+ .containsEntry(1, 43L);
+ }
+ }
+
+ @Test
+ public void testCheckpointDuringPendingRecoveryKeepsRetainedSplit() throws
Throwable {
+ int parallelism = 2;
+ String streamId = "stream";
+ String activeClusterId = "active-cluster";
+ String activeTopic = "active-topic";
+ DynamicKafkaSourceSplit activeSplit = createSplits(activeClusterId,
activeTopic, 1).get(0);
+ DynamicKafkaSourceSplit removedSplit =
+ createSplits("removed-cluster", "removed-topic", 1).get(0);
+ KafkaStream kafkaStream = createKafkaStream(streamId, activeClusterId,
activeTopic);
+ Properties properties = createGlobalModeProperties();
+ properties.setProperty(
+
DynamicKafkaSourceOptions.STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS.key(),
+ "60000");
+
+ DynamicKafkaSourceEnumState inWindowCheckpoint;
+ try (MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
+ new MockSplitEnumeratorContext<>(parallelism);
+ DynamicKafkaSourceEnumerator enumerator =
+ createEnumerator(
+ streamId,
+ new
MockKafkaMetadataService(Collections.singleton(kafkaStream)),
+ context,
+ properties,
+ createRestoredState(
+ kafkaStream,
+ activeClusterId,
+
Collections.singletonList(activeSplit)))) {
+ enumerator.start();
+ context.registerReader(
+ ReaderInfo.createReaderInfo(
+ 0, "location-0", List.of(activeSplit,
removedSplit)));
+ enumerator.addReader(0);
+ context.registerReader(
+ ReaderInfo.createReaderInfo(1, "location-1",
Collections.emptyList()));
+ enumerator.addReader(1);
+
+ assertThat(context.getSplitsAssignmentSequence()).isEmpty();
+ inWindowCheckpoint = enumerator.snapshotState(1L);
+ }
+
+ DynamicKafkaSourceEnumState restoredCheckpoint =
serdeRoundTrip(inWindowCheckpoint);
+ try (MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
+ new MockSplitEnumeratorContext<>(parallelism);
+ DynamicKafkaSourceEnumerator enumerator =
+ createEnumerator(
+ streamId,
+ new
MockKafkaMetadataService(Collections.singleton(kafkaStream)),
+ context,
+ properties,
+ restoredCheckpoint)) {
+ enumerator.start();
+ registerReadersWithoutReportedSplits(context, enumerator,
parallelism);
+ context.runNextOneTimeCallable();
+
+ assertAssignedSplitIdsExactly(context, List.of(activeSplit,
removedSplit));
+ DynamicKafkaSourceSplit retainedSplit =
+ collectAssignedSplits(context).stream()
+ .filter(split ->
split.splitId().equals(removedSplit.splitId()))
+ .findFirst()
+ .orElseThrow(AssertionError::new);
+ assertThat(retainedSplit.isRetained()).isTrue();
+ }
+ }
+
+ @Test
+ public void
testRestoredPendingReportedSplitsMergeWhenDownscaleCollapsesReaderIds()
+ throws Throwable {
+ int parallelism = 3;
+ String streamId = "stream";
+ String clusterId = "cluster-0";
+ String topic = "topic";
+ List<DynamicKafkaSourceSplit> activeSplits = createSplits(clusterId,
topic, 4);
+ DynamicKafkaSourceSplit removedClusterSplit =
+ createSplits("removed-cluster", "removed-topic", 1).get(0);
+ List<DynamicKafkaSourceSplit> reportedByReader4 =
+ new ArrayList<>(activeSplits.subList(2, 4));
+ reportedByReader4.add(removedClusterSplit);
+ List<DynamicKafkaSourceSplit> allReportedSplits = new
ArrayList<>(activeSplits);
+ allReportedSplits.add(removedClusterSplit);
+ KafkaStream kafkaStream = createKafkaStream(streamId, clusterId,
topic);
+ Properties properties = createGlobalModeProperties();
+ properties.setProperty(
+
DynamicKafkaSourceOptions.STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS.key(),
+ "60000");
+ // readers 1 and 4 from a parallelism-5 checkpoint both map to reader
floorMod(4, 3) = 1
+ Map<Integer, List<DynamicKafkaSourceSplit>> pendingSplitsByReader =
new HashMap<>();
+ pendingSplitsByReader.put(1, activeSplits.subList(0, 2));
+ pendingSplitsByReader.put(4, reportedByReader4);
+ DynamicKafkaSourceEnumState inWindowCheckpoint =
+ new DynamicKafkaSourceEnumState(
+ Collections.singleton(kafkaStream),
+ Collections.singletonMap(
+ clusterId,
+ new KafkaSourceEnumState(
+ unwrapSplits(activeSplits),
Collections.emptyList(), true)),
+ Collections.emptyMap(),
+ pendingSplitsByReader);
+
+ DynamicKafkaSourceEnumState restoredCheckpoint =
serdeRoundTrip(inWindowCheckpoint);
+ try (MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
+ new MockSplitEnumeratorContext<>(parallelism);
+ DynamicKafkaSourceEnumerator enumerator =
+ createEnumerator(
+ streamId,
+ new
MockKafkaMetadataService(Collections.singleton(kafkaStream)),
+ context,
+ properties,
+ restoredCheckpoint)) {
+ enumerator.start();
+ registerReadersWithoutReportedSplits(context, enumerator,
parallelism);
+ context.runNextOneTimeCallable();
+
+ Integer retainedSplitReader = null;
+ for (SplitsAssignment<DynamicKafkaSourceSplit> assignment :
+ context.getSplitsAssignmentSequence()) {
+ for (Map.Entry<Integer, List<DynamicKafkaSourceSplit>> entry :
+ assignment.assignment().entrySet()) {
+ assertThat(entry.getKey()).isBetween(0, parallelism - 1);
+ for (DynamicKafkaSourceSplit split : entry.getValue()) {
+ if
(split.splitId().equals(removedClusterSplit.splitId())) {
+ assertThat(split.isRetained()).isTrue();
+ retainedSplitReader = entry.getKey();
+ }
+ }
+ }
+ }
+ assertAssignedSplitIdsExactly(context, allReportedSplits);
+ // the retained split returns to its reporting reader, remapped as
floorMod(4, 3)
+ assertThat(retainedSplitReader).isEqualTo(1);
+ }
+ }
+
+ @Test
+ public void testRestoredPendingSplitsMergeWithFreshReport() throws
Throwable {
+ int parallelism = 2;
+ String streamId = "stream";
+ String clusterId = "cluster-0";
+ String topic = "topic";
+ // partition 0 is reported both as restored pending (offset 10) and
fresh (offset 42),
+ // partition 1 only as restored pending, partition 2 only as fresh
+ DynamicKafkaSourceSplit restoredOverlappingSplit =
+ new DynamicKafkaSourceSplit(
+ clusterId, new KafkaPartitionSplit(new
TopicPartition(topic, 0), 10));
+ DynamicKafkaSourceSplit restoredOnlySplit =
+ new DynamicKafkaSourceSplit(
+ clusterId, new KafkaPartitionSplit(new
TopicPartition(topic, 1), 7));
+ DynamicKafkaSourceSplit freshOverlappingSplit =
+ new DynamicKafkaSourceSplit(
+ clusterId, new KafkaPartitionSplit(new
TopicPartition(topic, 0), 42));
+ DynamicKafkaSourceSplit freshOnlySplit =
+ new DynamicKafkaSourceSplit(
+ clusterId, new KafkaPartitionSplit(new
TopicPartition(topic, 2), 5));
+ KafkaStream kafkaStream = createKafkaStream(streamId, clusterId,
topic);
+ DynamicKafkaSourceEnumState restoredCheckpoint =
+ serdeRoundTrip(
+ new DynamicKafkaSourceEnumState(
+ Collections.singleton(kafkaStream),
+ Collections.singletonMap(
+ clusterId,
+ new KafkaSourceEnumState(
+ unwrapSplits(
+ List.of(
+
restoredOverlappingSplit,
+
restoredOnlySplit,
+
freshOnlySplit)),
+ Collections.emptyList(),
+ true)),
+ Collections.emptyMap(),
+ Collections.singletonMap(
+ 1, List.of(restoredOverlappingSplit,
restoredOnlySplit))));
+
+ try (MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
+ new MockSplitEnumeratorContext<>(parallelism);
+ DynamicKafkaSourceEnumerator enumerator =
+ createEnumerator(
+ streamId,
+ new
MockKafkaMetadataService(Collections.singleton(kafkaStream)),
+ context,
+ createGlobalModeProperties(),
+ restoredCheckpoint)) {
+ enumerator.start();
+ context.registerReader(
+ ReaderInfo.createReaderInfo(0, "location-0",
Collections.emptyList()));
+ enumerator.addReader(0);
+ context.registerReader(
+ ReaderInfo.createReaderInfo(
+ 1, "location-1", List.of(freshOverlappingSplit,
freshOnlySplit)));
+ enumerator.addReader(1);
+ context.runNextOneTimeCallable();
+
+ assertAssignedSplitIdsExactly(
+ context, List.of(freshOverlappingSplit, restoredOnlySplit,
freshOnlySplit));
+ Map<Integer, Long> startingOffsetByPartition = new HashMap<>();
+ for (DynamicKafkaSourceSplit split :
collectAssignedSplits(context)) {
+ startingOffsetByPartition.put(
+ split.getKafkaPartitionSplit().getPartition(),
+ split.getKafkaPartitionSplit().getStartingOffset());
+ }
+ assertThat(startingOffsetByPartition)
+ .as("fresh report wins per split id, restored-only splits
keep their offsets")
+ .containsEntry(0, 42L)
+ .containsEntry(1, 7L)
+ .containsEntry(2, 5L);
+ }
+ }
+
+ @Test
+ public void
testChangedMetadataAfterInWindowCheckpointDropsRemovedTopicSplits()
+ throws Throwable {
+ int parallelism = 2;
+ String streamId = "stream";
+ String clusterId = "cluster-0";
+ String keptTopic = "kept-topic";
+ String removedTopic = "removed-topic";
+ List<DynamicKafkaSourceSplit> keptSplits = createSplits(clusterId,
keptTopic, 2);
+ List<DynamicKafkaSourceSplit> reportedSplits = new
ArrayList<>(keptSplits);
+ reportedSplits.addAll(createSplits(clusterId, removedTopic, 1));
+ KafkaStream restoredKafkaStream =
+ createKafkaStream(streamId, clusterId, Set.of(keptTopic,
removedTopic));
+ KafkaStream currentKafkaStream = createKafkaStream(streamId,
clusterId, keptTopic);
+ Properties properties = createGlobalModeProperties();
+ // retention enabled: the split is dropped because its topic was
removed while the
+ // cluster stayed active, not because retention is off
+ properties.setProperty(
+
DynamicKafkaSourceOptions.STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS.key(),
+ "60000");
+
+ DynamicKafkaSourceEnumState inWindowCheckpoint;
+ try (MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
+ new MockSplitEnumeratorContext<>(parallelism);
+ DynamicKafkaSourceEnumerator enumerator =
+ createEnumerator(
+ streamId,
+ new MockKafkaMetadataService(
+
Collections.singleton(restoredKafkaStream)),
+ context,
+ properties,
+ createRestoredState(
+ restoredKafkaStream, clusterId,
reportedSplits))) {
+ enumerator.start();
+ context.registerReader(ReaderInfo.createReaderInfo(0,
"location-0", reportedSplits));
+ enumerator.addReader(0);
+ context.registerReader(
+ ReaderInfo.createReaderInfo(1, "location-1",
Collections.emptyList()));
+ enumerator.addReader(1);
+
+ assertThat(context.getSplitsAssignmentSequence()).isEmpty();
+ inWindowCheckpoint = enumerator.snapshotState(1L);
+ }
+
+ DynamicKafkaSourceEnumState restoredCheckpoint =
serdeRoundTrip(inWindowCheckpoint);
+ try (MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
+ new MockSplitEnumeratorContext<>(parallelism);
+ DynamicKafkaSourceEnumerator enumerator =
+ createEnumerator(
+ streamId,
+ new MockKafkaMetadataService(
+
Collections.singleton(currentKafkaStream)),
+ context,
+ properties,
+ restoredCheckpoint)) {
+ enumerator.start();
+ registerReadersWithoutReportedSplits(context, enumerator,
parallelism);
+ context.runNextOneTimeCallable();
+
+ assertAssignedSplitIdsExactly(context, keptSplits);
+ }
+ }
+
+ @Test
+ public void
testPendingOnlyRestoredStateDefersMetadataUpdatesUntilReassignment()
+ throws Throwable {
+ int parallelism = 2;
+ String streamId = "stream";
+ String clusterId = "cluster-0";
+ String topic = "topic";
+ List<DynamicKafkaSourceSplit> splits = createSplits(clusterId, topic,
2);
+ KafkaStream kafkaStream = createKafkaStream(streamId, clusterId,
topic);
+ // the pending splits' cluster appears in neither the cluster nor the
retained states
+ DynamicKafkaSourceEnumState restoredCheckpoint =
+ serdeRoundTrip(
+ new DynamicKafkaSourceEnumState(
+ Collections.singleton(kafkaStream),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.singletonMap(0, splits)));
+
+ try (MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
+ new MockSplitEnumeratorContext<>(parallelism);
+ DynamicKafkaSourceEnumerator enumerator =
+ createEnumerator(
+ streamId,
+ new
MockKafkaMetadataService(Collections.singleton(kafkaStream)),
+ context,
+ createGlobalModeProperties(),
+ restoredCheckpoint)) {
+ enumerator.start();
+ registerReadersWithoutReportedSplits(context, enumerator,
parallelism);
+ enumerator.handleSourceEvent(0, new GetMetadataUpdateEvent());
+
+ // all readers are registered, but the restored pending splits are
not reassigned
+ // yet: the metadata update must stay deferred
+ assertThat(context.getSentSourceEvent().getOrDefault(0,
Collections.emptyList()))
+ .isEmpty();
+
+ context.runNextOneTimeCallable();
+
+ assertAssignedSplitIdsExactly(context, splits);
+ assertThat(context.getSentSourceEvent().get(0))
+ .hasSize(1)
+ .allMatch(MetadataUpdateEvent.class::isInstance);
+ }
+ }
+
+ private static void registerReadersWithoutReportedSplits(
+ MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context,
+ DynamicKafkaSourceEnumerator enumerator,
+ int parallelism) {
+ for (int reader = 0; reader < parallelism; reader++) {
+ context.registerReader(
+ ReaderInfo.createReaderInfo(
+ reader, "location-" + reader,
Collections.emptyList()));
+ enumerator.addReader(reader);
+ }
+ }
+
+ private static DynamicKafkaSourceEnumState
serdeRoundTrip(DynamicKafkaSourceEnumState state)
+ throws Exception {
+ DynamicKafkaSourceEnumStateSerializer serializer =
+ new DynamicKafkaSourceEnumStateSerializer();
+ return serializer.deserialize(serializer.getVersion(),
serializer.serialize(state));
+ }
+
+ /**
+ * Asserts the assigned splits match the expected splits by id, including
multiplicity, so a
+ * duplicate assignment of the same split cannot go unnoticed.
+ */
+ private static void assertAssignedSplitIdsExactly(
+ MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context,
+ List<DynamicKafkaSourceSplit> expectedSplits) {
+ List<String> assignedSplitIds = new ArrayList<>();
+ for (DynamicKafkaSourceSplit split : collectAssignedSplits(context)) {
+ assignedSplitIds.add(split.splitId());
+ }
+ List<String> expectedSplitIds = new ArrayList<>();
+ for (DynamicKafkaSourceSplit split : expectedSplits) {
+ expectedSplitIds.add(split.splitId());
+ }
+
assertThat(assignedSplitIds).containsExactlyInAnyOrderElementsOf(expectedSplitIds);
+ }
+
+ private static List<DynamicKafkaSourceSplit> collectAssignedSplits(
+ MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context) {
+ List<DynamicKafkaSourceSplit> assignedSplits = new ArrayList<>();
+ for (SplitsAssignment<DynamicKafkaSourceSplit> assignment :
+ context.getSplitsAssignmentSequence()) {
+ for (List<DynamicKafkaSourceSplit> readerSplits :
assignment.assignment().values()) {
+ assignedSplits.addAll(readerSplits);
+ }
+ }
+ return assignedSplits;
+ }
+
private static DynamicKafkaSourceEnumState createRestoredState(
KafkaStream kafkaStream, String clusterId,
List<DynamicKafkaSourceSplit> activeSplits) {
return new DynamicKafkaSourceEnumState(
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 e6456ed4..cf3d5376 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
@@ -27,6 +27,7 @@ import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.NavigableMap;
import static org.assertj.core.api.Assertions.assertThat;
@@ -97,19 +98,44 @@ class ReaderRecoveryGateTest {
}
@Test
- void testRepeatedReportForSameReaderReplacesPreviousReport() {
+ void testRepeatedReportMergesSplitsAndPrefersLatestOffsets() {
ReaderRecoveryGate gate = new ReaderRecoveryGate(false);
DynamicKafkaSourceSplit firstReport = split("topic", 0);
DynamicKafkaSourceSplit secondReport = split("topic", 1);
- gate.recordReportedSplits(1, Collections.singletonList(firstReport));
- gate.recordReportedSplits(1, Collections.singletonList(secondReport));
+ DynamicKafkaSourceSplit updatedSplit =
+ new DynamicKafkaSourceSplit(
+ "cluster0", new KafkaPartitionSplit(new
TopicPartition("topic", 0), 42L));
+ DynamicKafkaSourceSplit previousOnlySplit = split("topic", 2);
+ gate.recordReportedSplits(1, Arrays.asList(firstReport,
previousOnlySplit));
+ gate.recordReportedSplits(1, Arrays.asList(updatedSplit,
secondReport));
- // A reader that registers again keeps one entry holding its latest
report, so the
- // duplicate-owner check in reassignReportedSplits is not tripped by
its own re-report.
NavigableMap<Integer, List<DynamicKafkaSourceSplit>> drained =
gate.drainReportedSplits();
assertThat(drained.keySet()).containsExactly(1);
- assertThat(drained.get(1)).containsExactly(secondReport);
+ assertThat(drained.get(1))
+ .containsExactlyInAnyOrder(updatedSplit, previousOnlySplit,
secondReport);
+ }
+
+ @Test
+ void testSnapshotReportedSplitsIsIndependentAndDoesNotDrain() {
+ ReaderRecoveryGate gate = new ReaderRecoveryGate(false);
+ DynamicKafkaSourceSplit firstReport = split("topic", 0);
+ DynamicKafkaSourceSplit secondReport = split("topic", 1);
+ gate.recordReportedSplits(1, Collections.singletonList(firstReport));
+
+ Map<Integer, List<DynamicKafkaSourceSplit>> snapshot =
gate.snapshotReportedSplits();
+ assertThat(gate.hasReportedSplits()).isTrue();
+ snapshot.get(1).clear();
+ snapshot.clear();
+
assertThat(gate.snapshotReportedSplits().get(1)).containsExactly(firstReport);
+
+ snapshot = gate.snapshotReportedSplits();
+ gate.recordReportedSplits(1, Collections.singletonList(secondReport));
+ assertThat(gate.drainReportedSplits().get(1))
+ .containsExactlyInAnyOrder(firstReport, secondReport);
+ assertThat(gate.snapshotReportedSplits()).isEmpty();
+ assertThat(snapshot).containsOnlyKeys(1);
+ assertThat(snapshot.get(1)).containsExactly(firstReport);
}
@Test
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/testutils/DynamicKafkaSourceEnumStateTestUtils.java
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/testutils/DynamicKafkaSourceEnumStateTestUtils.java
index f6038784..08feb30b 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/testutils/DynamicKafkaSourceEnumStateTestUtils.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/testutils/DynamicKafkaSourceEnumStateTestUtils.java
@@ -18,6 +18,7 @@
package org.apache.flink.connector.kafka.dynamic.source.testutils;
+import org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumState;
import
org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumStateSerializer;
import java.io.ByteArrayOutputStream;
@@ -50,4 +51,67 @@ public final class DynamicKafkaSourceEnumStateTestUtils {
return baos.toByteArray();
}
}
+
+ public static byte[] serializeV2State(
+ String streamId,
+ String clusterId,
+ Set<String> topics,
+ String bootstrapServers,
+ KafkaSourceEnumState clusterState)
+ throws IOException {
+ KafkaSourceEnumStateSerializer kafkaSourceEnumStateSerializer =
+ new KafkaSourceEnumStateSerializer();
+ try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(baos)) {
+ // kafka streams in the V2 layout
+ out.writeInt(1);
+ out.writeUTF(streamId);
+ out.writeInt(1);
+ out.writeUTF(clusterId);
+ out.writeInt(topics.size());
+ for (String topic : topics) {
+ out.writeUTF(topic);
+ }
+ out.writeUTF(bootstrapServers);
+ out.writeBoolean(false); // no starting offsets initializer
+ out.writeBoolean(false); // no stopping offsets initializer
+ out.writeInt(kafkaSourceEnumStateSerializer.getVersion());
+ // cluster enumerator states
+ out.writeInt(1);
+ out.writeUTF(clusterId);
+
writeLengthPrefixed(kafkaSourceEnumStateSerializer.serialize(clusterState),
out);
+ return baos.toByteArray();
+ }
+ }
+
+ public static byte[] serializeV3State(
+ String streamId,
+ String clusterId,
+ Set<String> topics,
+ String bootstrapServers,
+ KafkaSourceEnumState clusterState,
+ String retainedClusterId,
+ long retainedUntilMs,
+ KafkaSourceEnumState retainedClusterState)
+ throws IOException {
+ KafkaSourceEnumStateSerializer kafkaSourceEnumStateSerializer =
+ new KafkaSourceEnumStateSerializer();
+ try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(baos)) {
+ out.write(
+ serializeV2State(streamId, clusterId, topics,
bootstrapServers, clusterState));
+ // retained cluster enumerator states (new in V3)
+ out.writeInt(1);
+ out.writeUTF(retainedClusterId);
+ out.writeLong(retainedUntilMs);
+ writeLengthPrefixed(
+
kafkaSourceEnumStateSerializer.serialize(retainedClusterState), out);
+ return baos.toByteArray();
+ }
+ }
+
+ private static void writeLengthPrefixed(byte[] bytes, DataOutputStream
out) throws IOException {
+ out.writeInt(bytes.length);
+ out.write(bytes);
+ }
}
