This is an automated email from the ASF dual-hosted git repository.
bowenli86 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 cb40de0e [FLINK-39837] Retain removed dynamic Kafka cluster offsets
(#264)
cb40de0e is described below
commit cb40de0ef41aadbc735dfae31b0d23b48e58f9f3
Author: lnbest0707 <[email protected]>
AuthorDate: Thu Jun 18 08:40:52 2026 -0700
[FLINK-39837] Retain removed dynamic Kafka cluster offsets (#264)
Retain removed DynamicKafkaSource cluster split offsets and enumerator
state in checkpoints for a configurable retention window so a temporarily
removed cluster can resume from checkpointed offsets when it is added back.
---
.../docs/connectors/datastream/dynamic-kafka.md | 12 +
.../docs/connectors/datastream/dynamic-kafka.md | 12 +
.../dynamic/source/DynamicKafkaSourceOptions.java | 27 ++
.../enumerator/DynamicKafkaSourceEnumState.java | 34 +++
.../DynamicKafkaSourceEnumStateSerializer.java | 98 ++++++--
.../enumerator/DynamicKafkaSourceEnumerator.java | 96 +++++--
.../source/reader/DynamicKafkaSourceReader.java | 81 +++++-
.../source/split/DynamicKafkaSourceSplit.java | 39 ++-
.../split/DynamicKafkaSourceSplitSerializer.java | 23 +-
.../kafka/table/DynamicKafkaTableFactory.java | 10 +
.../dynamic/source/DynamicKafkaSourceITTest.java | 278 +++++++++++++++++++++
.../DynamicKafkaSourceEnumStateSerializerTest.java | 11 +-
.../DynamicKafkaSourceEnumeratorTest.java | 157 ++++++++++++
.../reader/DynamicKafkaSourceReaderTest.java | 66 +++++
.../DynamicKafkaSourceSplitSerializerTest.java | 49 +++-
.../kafka/table/DynamicKafkaTableFactoryTest.java | 20 ++
16 files changed, 950 insertions(+), 63 deletions(-)
diff --git a/docs/content.zh/docs/connectors/datastream/dynamic-kafka.md
b/docs/content.zh/docs/connectors/datastream/dynamic-kafka.md
index cdff12f4..8f4adc6d 100644
--- a/docs/content.zh/docs/connectors/datastream/dynamic-kafka.md
+++ b/docs/content.zh/docs/connectors/datastream/dynamic-kafka.md
@@ -229,6 +229,11 @@ swap from one cluster to the new cluster when the service
makes that change in t
Cluster metadata 可以包含每个集群的起始/停止 offsets initializer，用于覆盖全局 builder 配置。
+默认情况下，从 metadata 中移除集群后，后续 checkpoint 也会移除该集群的 split offset。
+如果希望在之后重新加入集群或恢复作业时继续使用这些 offset，请将
+`stream-metadata-removed-cluster-retention-ms` 设置为正数。例如，`604800000`
+会将已移除集群的状态保留七天，之后 source 将不再把它写入 checkpoint。
+
### Additional Properties
There are configuration options in DynamicKafkaSourceOptions that can be
configured in the properties through the builder:
<table class="table table-bordered">
@@ -256,6 +261,13 @@ There are configuration options in
DynamicKafkaSourceOptions that can be configu
<td>Integer</td>
<td>The number of consecutive failures before letting the exception from
Kafka metadata service discovery trigger jobmanager failure and global
failover. The default is one to at least catch startup failures.</td>
</tr>
+ <tr>
+ <td><h5>stream-metadata-removed-cluster-retention-ms</h5></td>
+ <td>required</td>
+ <td style="word-wrap: break-word;">0</td>
+ <td>Long</td>
+ <td>已移除 Kafka 集群的 split offset 和 enumerator 状态继续写入 checkpoint
的时长，单位为毫秒。零值会禁用保留。</td>
+ </tr>
<tr>
<td><h5>stream-enumerator-mode</h5></td>
<td>required</td>
diff --git a/docs/content/docs/connectors/datastream/dynamic-kafka.md
b/docs/content/docs/connectors/datastream/dynamic-kafka.md
index 60b2e504..8df05afb 100644
--- a/docs/content/docs/connectors/datastream/dynamic-kafka.md
+++ b/docs/content/docs/connectors/datastream/dynamic-kafka.md
@@ -238,6 +238,11 @@ swap from one cluster to the new cluster when the service
makes that change in t
Cluster metadata can optionally carry per-cluster starting and stopping
offsets initializers. These
override the global builder configuration for the affected cluster.
+By default, metadata removal also removes that cluster's split offsets from
subsequent checkpoints.
+To keep removed cluster offsets available for a later re-add or restore, set
+`stream-metadata-removed-cluster-retention-ms` to a positive duration. For
example,
+`604800000` retains removed cluster state for seven days before the source
stops checkpointing it.
+
### Additional Properties
There are configuration options in DynamicKafkaSourceOptions that can be
configured in the properties through the builder:
<table class="table table-bordered">
@@ -265,6 +270,13 @@ There are configuration options in
DynamicKafkaSourceOptions that can be configu
<td>Integer</td>
<td>The number of consecutive failures before letting the exception from
Kafka metadata service discovery trigger jobmanager failure and global
failover. The default is one to at least catch startup failures.</td>
</tr>
+ <tr>
+ <td><h5>stream-metadata-removed-cluster-retention-ms</h5></td>
+ <td>required</td>
+ <td style="word-wrap: break-word;">0</td>
+ <td>Long</td>
+ <td>The duration in milliseconds that removed Kafka cluster split
offsets and enumerator state stay in checkpoints. Zero disables retention.</td>
+ </tr>
<tr>
<td><h5>stream-enumerator-mode</h5></td>
<td>required</td>
diff --git
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/DynamicKafkaSourceOptions.java
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/DynamicKafkaSourceOptions.java
index 56c6d201..845cb7f5 100644
---
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/DynamicKafkaSourceOptions.java
+++
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/DynamicKafkaSourceOptions.java
@@ -62,6 +62,14 @@ public class DynamicKafkaSourceOptions {
+ "trigger jobmanager failure and global
failover. The default is one to at least catch startup "
+ "failures.");
+ public static final ConfigOption<Long>
STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS =
+ ConfigOptions.key("stream-metadata-removed-cluster-retention-ms")
+ .longType()
+ .defaultValue(0L)
+ .withDescription(
+ "The duration in milliseconds that removed Kafka
cluster split offsets "
+ + "and enumerator state stay in
checkpoints. Zero disables retention.");
+
public static final ConfigOption<String> STREAM_ENUMERATOR_MODE =
ConfigOptions.key("stream-enumerator-mode")
.stringType()
@@ -108,4 +116,23 @@ public class DynamicKafkaSourceOptions {
}
});
}
+
+ @Internal
+ public static long getRemovedClusterStateRetentionMs(Properties props) {
+ long retentionMs =
+ getOption(props, STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS,
Long::parseLong);
+ if (retentionMs < 0) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Invalid %s='%d'. The value must be non-negative.",
+
STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS.key(), retentionMs));
+ }
+
+ return retentionMs;
+ }
+
+ @Internal
+ public static void removeRemovedClusterRetentionOption(Properties props) {
+ props.remove(STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS.key());
+ }
}
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 0dd0a774..bb474d8c 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
@@ -35,17 +35,27 @@ import java.util.Set;
public class DynamicKafkaSourceEnumState {
private final Set<KafkaStream> kafkaStreams;
private final Map<String, KafkaSourceEnumState> clusterEnumeratorStates;
+ private final Map<String, RetainedClusterState>
retainedClusterEnumeratorStates;
public DynamicKafkaSourceEnumState() {
this.kafkaStreams = new HashSet<>();
this.clusterEnumeratorStates = new HashMap<>();
+ this.retainedClusterEnumeratorStates = new HashMap<>();
}
public DynamicKafkaSourceEnumState(
Set<KafkaStream> kafkaStreams,
Map<String, KafkaSourceEnumState> clusterEnumeratorStates) {
+ this(kafkaStreams, clusterEnumeratorStates, new HashMap<>());
+ }
+
+ public DynamicKafkaSourceEnumState(
+ Set<KafkaStream> kafkaStreams,
+ Map<String, KafkaSourceEnumState> clusterEnumeratorStates,
+ Map<String, RetainedClusterState> retainedClusterEnumeratorStates)
{
this.kafkaStreams = kafkaStreams;
this.clusterEnumeratorStates = clusterEnumeratorStates;
+ this.retainedClusterEnumeratorStates = retainedClusterEnumeratorStates;
}
public Set<KafkaStream> getKafkaStreams() {
@@ -55,4 +65,28 @@ public class DynamicKafkaSourceEnumState {
public Map<String, KafkaSourceEnumState> getClusterEnumeratorStates() {
return clusterEnumeratorStates;
}
+
+ public Map<String, RetainedClusterState>
getRetainedClusterEnumeratorStates() {
+ return retainedClusterEnumeratorStates;
+ }
+
+ /** Kafka enumerator state that stays checkpointed after its cluster
becomes inactive. */
+ public static class RetainedClusterState {
+ private final KafkaSourceEnumState kafkaSourceEnumState;
+ private final long retainedUntilMs;
+
+ public RetainedClusterState(
+ KafkaSourceEnumState kafkaSourceEnumState, long
retainedUntilMs) {
+ this.kafkaSourceEnumState = kafkaSourceEnumState;
+ this.retainedUntilMs = retainedUntilMs;
+ }
+
+ public KafkaSourceEnumState getKafkaSourceEnumState() {
+ return kafkaSourceEnumState;
+ }
+
+ public long getRetainedUntilMs() {
+ return retainedUntilMs;
+ }
+ }
}
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 b7f0b49a..1cec2a18 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
@@ -50,6 +50,7 @@ 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 final KafkaSourceEnumStateSerializer
kafkaSourceEnumStateSerializer;
@@ -59,7 +60,7 @@ public class DynamicKafkaSourceEnumStateSerializer
@Override
public int getVersion() {
- return VERSION_2;
+ return VERSION_3;
}
@Override
@@ -74,19 +75,17 @@ public class DynamicKafkaSourceEnumStateSerializer
state.getClusterEnumeratorStates();
out.writeInt(kafkaSourceEnumStateSerializer.getVersion());
- // write sub enumerator states
- out.writeInt(clusterEnumeratorStates.size());
- for (Map.Entry<String, KafkaSourceEnumState>
clusterEnumeratorState :
- clusterEnumeratorStates.entrySet()) {
- String kafkaClusterId = clusterEnumeratorState.getKey();
- out.writeUTF(kafkaClusterId);
- byte[] bytes =
-
kafkaSourceEnumStateSerializer.serialize(clusterEnumeratorState.getValue());
- // we need to know the exact size of the byte array since
- // KafkaSourceEnumStateSerializer
- // will throw exception if there are leftover unread bytes in
deserialization.
- out.writeInt(bytes.length);
- out.write(bytes);
+ writeClusterEnumeratorStates(clusterEnumeratorStates, out);
+
+ Map<String, DynamicKafkaSourceEnumState.RetainedClusterState>
+ retainedClusterEnumeratorStates =
state.getRetainedClusterEnumeratorStates();
+ out.writeInt(retainedClusterEnumeratorStates.size());
+ for (Map.Entry<String,
DynamicKafkaSourceEnumState.RetainedClusterState>
+ retainedClusterEnumeratorState :
retainedClusterEnumeratorStates.entrySet()) {
+ out.writeUTF(retainedClusterEnumeratorState.getKey());
+
out.writeLong(retainedClusterEnumeratorState.getValue().getRetainedUntilMs());
+ writeKafkaSourceEnumState(
+
retainedClusterEnumeratorState.getValue().getKafkaSourceEnumState(), out);
}
return baos.toByteArray();
@@ -96,7 +95,7 @@ public class DynamicKafkaSourceEnumStateSerializer
@Override
public DynamicKafkaSourceEnumState deserialize(int version, byte[]
serialized)
throws IOException {
- if (version != VERSION_1 && version != VERSION_2) {
+ if (version != VERSION_1 && version != VERSION_2 && version !=
VERSION_3) {
throw new IOException(
String.format(
"The bytes are serialized with version %d, "
@@ -112,21 +111,70 @@ public class DynamicKafkaSourceEnumStateSerializer
Map<String, KafkaSourceEnumState> clusterEnumeratorStates = new
HashMap<>();
int kafkaSourceEnumStateSerializerVersion = in.readInt();
- int clusterEnumeratorStateMapSize = in.readInt();
- for (int i = 0; i < clusterEnumeratorStateMapSize; i++) {
- String kafkaClusterId = in.readUTF();
- int byteArraySize = in.readInt();
- KafkaSourceEnumState kafkaSourceEnumState =
- kafkaSourceEnumStateSerializer.deserialize(
- kafkaSourceEnumStateSerializerVersion,
- readNBytes(in, byteArraySize));
- clusterEnumeratorStates.put(kafkaClusterId,
kafkaSourceEnumState);
+ readClusterEnumeratorStates(
+ in, kafkaSourceEnumStateSerializerVersion,
clusterEnumeratorStates);
+
+ Map<String, DynamicKafkaSourceEnumState.RetainedClusterState>
+ retainedClusterEnumeratorStates = new HashMap<>();
+ if (version == VERSION_3) {
+ int retainedClusterEnumeratorStateMapSize = in.readInt();
+ for (int i = 0; i < retainedClusterEnumeratorStateMapSize;
i++) {
+ String kafkaClusterId = in.readUTF();
+ long retainedUntilMs = in.readLong();
+ KafkaSourceEnumState kafkaSourceEnumState =
+ readKafkaSourceEnumState(in,
kafkaSourceEnumStateSerializerVersion);
+ retainedClusterEnumeratorStates.put(
+ kafkaClusterId,
+ new
DynamicKafkaSourceEnumState.RetainedClusterState(
+ kafkaSourceEnumState, retainedUntilMs));
+ }
}
- return new DynamicKafkaSourceEnumState(kafkaStreams,
clusterEnumeratorStates);
+ return new DynamicKafkaSourceEnumState(
+ kafkaStreams, clusterEnumeratorStates,
retainedClusterEnumeratorStates);
+ }
+ }
+
+ private void writeClusterEnumeratorStates(
+ Map<String, KafkaSourceEnumState> clusterEnumeratorStates,
DataOutputStream out)
+ throws IOException {
+ out.writeInt(clusterEnumeratorStates.size());
+ for (Map.Entry<String, KafkaSourceEnumState> clusterEnumeratorState :
+ clusterEnumeratorStates.entrySet()) {
+ out.writeUTF(clusterEnumeratorState.getKey());
+ writeKafkaSourceEnumState(clusterEnumeratorState.getValue(), out);
}
}
+ private void readClusterEnumeratorStates(
+ DataInputStream in,
+ int kafkaSourceEnumStateSerializerVersion,
+ Map<String, KafkaSourceEnumState> clusterEnumeratorStates)
+ throws IOException {
+ int clusterEnumeratorStateMapSize = in.readInt();
+ for (int i = 0; i < clusterEnumeratorStateMapSize; i++) {
+ String kafkaClusterId = in.readUTF();
+ clusterEnumeratorStates.put(
+ kafkaClusterId,
+ readKafkaSourceEnumState(in,
kafkaSourceEnumStateSerializerVersion));
+ }
+ }
+
+ private void writeKafkaSourceEnumState(
+ KafkaSourceEnumState kafkaSourceEnumState, DataOutputStream out)
throws IOException {
+ byte[] bytes =
kafkaSourceEnumStateSerializer.serialize(kafkaSourceEnumState);
+ // KafkaSourceEnumStateSerializer rejects leftover unread bytes on
deserialization.
+ out.writeInt(bytes.length);
+ out.write(bytes);
+ }
+
+ private KafkaSourceEnumState readKafkaSourceEnumState(
+ DataInputStream in, int kafkaSourceEnumStateSerializerVersion)
throws IOException {
+ int byteArraySize = in.readInt();
+ return kafkaSourceEnumStateSerializer.deserialize(
+ kafkaSourceEnumStateSerializerVersion, readNBytes(in,
byteArraySize));
+ }
+
private void serializeV2(Set<KafkaStream> kafkaStreams, DataOutputStream
out)
throws IOException {
out.writeInt(kafkaStreams.size());
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 ba20fa98..cc3a34f7 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
@@ -94,11 +94,14 @@ public class DynamicKafkaSourceEnumerator
// options
private final long kafkaMetadataServiceDiscoveryIntervalMs;
private final int kafkaMetadataServiceDiscoveryFailureThreshold;
+ private final long removedClusterStateRetentionMs;
// state
private int kafkaMetadataServiceDiscoveryFailureCount;
private Map<String, Set<String>> latestClusterTopicsMap;
private Set<KafkaStream> latestKafkaStreams;
+ private Map<String, DynamicKafkaSourceEnumState.RetainedClusterState>
+ retainedClusterEnumeratorStates;
private boolean firstDiscoveryComplete;
public DynamicKafkaSourceEnumerator(
@@ -154,6 +157,8 @@ public class DynamicKafkaSourceEnumerator
properties,
DynamicKafkaSourceOptions.STREAM_METADATA_DISCOVERY_FAILURE_THRESHOLD,
Integer::parseInt);
+ this.removedClusterStateRetentionMs =
+
DynamicKafkaSourceOptions.getRemovedClusterStateRetentionMs(properties);
this.kafkaMetadataServiceDiscoveryFailureCount = 0;
this.firstDiscoveryComplete = false;
@@ -173,6 +178,9 @@ public class DynamicKafkaSourceEnumerator
this.latestKafkaStreams =
refreshRestoredClusterPropertiesFromMetadataService(this.latestKafkaStreams);
}
+ this.retainedClusterEnumeratorStates =
+ new
HashMap<>(dynamicKafkaSourceEnumState.getRetainedClusterEnumeratorStates());
+ pruneExpiredRetainedClusterEnumeratorStates();
Map<String, Properties> clusterProperties = new HashMap<>();
Map<String, OffsetsInitializer> clusterStartingOffsets = new
HashMap<>();
@@ -342,6 +350,7 @@ public class DynamicKafkaSourceEnumerator
firstDiscoveryComplete = true;
Set<KafkaStream> handledFetchKafkaStreams =
handleFetchSubscribedStreamsError(fetchedKafkaStreams, t);
+ pruneExpiredRetainedClusterEnumeratorStates();
Map<String, Set<String>> newClustersTopicsMap = new HashMap<>();
Map<String, Properties> clusterProperties = new HashMap<>();
@@ -395,29 +404,34 @@ public class DynamicKafkaSourceEnumerator
latestKafkaStreams = handledFetchKafkaStreams;
sendMetadataUpdateEventToAvailableReaders();
+ retainRemovedClusterEnumeratorStates(
+ dynamicKafkaSourceEnumState.getClusterEnumeratorStates(),
+ latestClusterTopicsMap.keySet());
+
// create enumerators
Set<String> activeSplitIds = new HashSet<>();
for (Entry<String, Set<String>> activeClusterTopics :
latestClusterTopicsMap.entrySet()) {
+ String kafkaClusterId = activeClusterTopics.getKey();
KafkaSourceEnumState kafkaSourceEnumState =
- dynamicKafkaSourceEnumState
- .getClusterEnumeratorStates()
- .get(activeClusterTopics.getKey());
+
dynamicKafkaSourceEnumState.getClusterEnumeratorStates().get(kafkaClusterId);
+ if (kafkaSourceEnumState == null) {
+ DynamicKafkaSourceEnumState.RetainedClusterState
retainedClusterState =
+ retainedClusterEnumeratorStates.remove(kafkaClusterId);
+ if (retainedClusterState != null) {
+ kafkaSourceEnumState =
retainedClusterState.getKafkaSourceEnumState();
+ }
+ } else {
+ retainedClusterEnumeratorStates.remove(kafkaClusterId);
+ }
final KafkaSourceEnumState newKafkaSourceEnumState;
if (kafkaSourceEnumState != null) {
- final Set<String> activeTopics =
activeClusterTopics.getValue();
-
- // filter out removed topics
Set<SplitAndAssignmentStatus> partitions =
- kafkaSourceEnumState.splits().stream()
- .filter(tp ->
activeTopics.contains(tp.split().getTopic()))
- .collect(Collectors.toSet());
+ filterStateByTopics(kafkaSourceEnumState,
activeClusterTopics.getValue());
partitions.forEach(
splitStatus ->
activeSplitIds.add(
- toDynamicSplitId(
- activeClusterTopics.getKey(),
- splitStatus.split())));
+ toDynamicSplitId(kafkaClusterId,
splitStatus.split())));
newKafkaSourceEnumState =
new KafkaSourceEnumState(
@@ -426,15 +440,17 @@ public class DynamicKafkaSourceEnumerator
newKafkaSourceEnumState = new
KafkaSourceEnumState(Collections.emptySet(), false);
}
- // restarts enumerator from state using only the active topic
partitions, to avoid
- // sending duplicate splits from enumerator
+ // Restart the enumerator from the active topic partitions already
known in state. The
+ // reader restores those splits from its own checkpointed offsets
during metadata
+ // reconciliation, so the enumerator must not send them again as
newly discovered
+ // splits.
createEnumeratorWithAssignedTopicPartitions(
- activeClusterTopics.getKey(),
+ kafkaClusterId,
activeClusterTopics.getValue(),
newKafkaSourceEnumState,
- clusterProperties.get(activeClusterTopics.getKey()),
- clusterStartingOffsets.get(activeClusterTopics.getKey()),
- clusterStoppingOffsets.get(activeClusterTopics.getKey()));
+ clusterProperties.get(kafkaClusterId),
+ clusterStartingOffsets.get(kafkaClusterId),
+ clusterStoppingOffsets.get(kafkaClusterId));
}
splitAssignmentStrategy.onMetadataRefresh(activeSplitIds);
@@ -516,6 +532,7 @@ public class DynamicKafkaSourceEnumerator
Properties consumerProps = new Properties();
KafkaPropertiesUtil.copyProperties(fetchedProperties, consumerProps);
KafkaPropertiesUtil.copyProperties(properties, consumerProps);
+
DynamicKafkaSourceOptions.removeRemovedClusterRetentionOption(consumerProps);
KafkaPropertiesUtil.setClientIdPrefix(consumerProps, kafkaClusterId);
consumerProps.setProperty(
ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
@@ -642,6 +659,7 @@ public class DynamicKafkaSourceEnumerator
*/
@Override
public DynamicKafkaSourceEnumState snapshotState(long checkpointId) throws
Exception {
+ pruneExpiredRetainedClusterEnumeratorStates();
Map<String, KafkaSourceEnumState> subEnumeratorStateByCluster = new
HashMap<>();
boolean isCheckpointSnapshot = checkpointId >= 0;
@@ -664,7 +682,47 @@ public class DynamicKafkaSourceEnumerator
}
}
- return new DynamicKafkaSourceEnumState(latestKafkaStreams,
subEnumeratorStateByCluster);
+ return new DynamicKafkaSourceEnumState(
+ latestKafkaStreams,
+ subEnumeratorStateByCluster,
+ new HashMap<>(retainedClusterEnumeratorStates));
+ }
+
+ private void retainRemovedClusterEnumeratorStates(
+ Map<String, KafkaSourceEnumState> activeClusterEnumeratorStates,
+ Set<String> activeKafkaClusterIds) {
+ if (removedClusterStateRetentionMs <= 0) {
+ return;
+ }
+
+ long retainedUntilMs = System.currentTimeMillis() +
removedClusterStateRetentionMs;
+ activeClusterEnumeratorStates.entrySet().stream()
+ .filter(entry ->
!activeKafkaClusterIds.contains(entry.getKey()))
+ .forEach(
+ entry ->
+ retainedClusterEnumeratorStates.put(
+ entry.getKey(),
+ new
DynamicKafkaSourceEnumState.RetainedClusterState(
+ entry.getValue(),
retainedUntilMs)));
+ }
+
+ private void pruneExpiredRetainedClusterEnumeratorStates() {
+ if (removedClusterStateRetentionMs <= 0) {
+ retainedClusterEnumeratorStates.clear();
+ return;
+ }
+
+ long currentTimeMillis = System.currentTimeMillis();
+ retainedClusterEnumeratorStates
+ .entrySet()
+ .removeIf(entry -> entry.getValue().getRetainedUntilMs() <=
currentTimeMillis);
+ }
+
+ private Set<SplitAndAssignmentStatus> filterStateByTopics(
+ KafkaSourceEnumState kafkaSourceEnumState, Set<String>
activeTopics) {
+ return kafkaSourceEnumState.splits().stream()
+ .filter(splitStatus ->
activeTopics.contains(splitStatus.split().getTopic()))
+ .collect(Collectors.toSet());
}
private static String
summarizeSplitOffsets(Collection<KafkaPartitionSplit> splits) {
diff --git
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java
index 3e4d3cf6..f5103392 100644
---
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java
+++
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java
@@ -30,6 +30,7 @@ import
org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
import
org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
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.DynamicKafkaSourceOptions;
import org.apache.flink.connector.kafka.dynamic.source.GetMetadataUpdateEvent;
import org.apache.flink.connector.kafka.dynamic.source.MetadataUpdateEvent;
import
org.apache.flink.connector.kafka.dynamic.source.metrics.KafkaClusterMetricGroup;
@@ -94,6 +95,8 @@ public class DynamicKafkaSourceReader<T> implements
SourceReader<T, DynamicKafka
private final Map<String, Properties> clustersProperties;
private final List<DynamicKafkaSourceSplit> pendingSplits;
private final Set<String> pendingSplitOutputReleases;
+ private final List<DynamicKafkaSourceSplit> retainedSplits;
+ private final long removedClusterStateRetentionMs;
private MultipleFuturesAvailabilityHelper availabilityHelper;
private int availabilityHelperSize;
@@ -117,6 +120,9 @@ public class DynamicKafkaSourceReader<T> implements
SourceReader<T, DynamicKafka
.addGroup(KafkaClusterMetricGroup.DYNAMIC_KAFKA_SOURCE_METRIC_GROUP);
this.kafkaClusterMetricGroupManager = new
KafkaClusterMetricGroupManager();
this.pendingSplits = new ArrayList<>();
+ this.retainedSplits = new ArrayList<>();
+ this.removedClusterStateRetentionMs =
+
DynamicKafkaSourceOptions.getRemovedClusterStateRetentionMs(properties);
this.availabilityHelper =
new
MultipleFuturesAvailabilityHelper(this.availabilityHelperSize = 0);
this.isNoMoreSplits = false;
@@ -249,6 +255,7 @@ public class DynamicKafkaSourceReader<T> implements
SourceReader<T, DynamicKafka
"Received source event {}: subtask={}",
sourceEvent,
readerContext.getIndexOfSubtask());
+ pruneExpiredRetainedSplits();
Set<KafkaStream> newKafkaStreams = ((MetadataUpdateEvent)
sourceEvent).getKafkaStreams();
Map<String, Set<String>> newClustersAndTopics = new HashMap<>();
Map<String, Properties> newClustersProperties = new HashMap<>();
@@ -285,6 +292,7 @@ public class DynamicKafkaSourceReader<T> implements
SourceReader<T, DynamicKafka
currentSplitState);
Map<String, Set<String>> currentMetadataFromState = new HashMap<>();
Map<String, List<KafkaPartitionSplit>> filteredNewClusterSplitStateMap
= new HashMap<>();
+ long retainedUntilMs = System.currentTimeMillis() +
removedClusterStateRetentionMs;
// the data structures above
for (DynamicKafkaSourceSplit split : currentSplitState) {
@@ -301,9 +309,13 @@ public class DynamicKafkaSourceReader<T> implements
SourceReader<T, DynamicKafka
.add(split);
} else {
releaseOrDeferSplitOutput(split.splitId());
+ if (shouldRetainSplit(split, newClustersAndTopics)) {
+ retainedSplits.add(split.retainUntil(retainedUntilMs));
+ }
logger.info("Skipping outdated split due to metadata changes:
{}", split);
}
}
+ reactivateRetainedSplits(newClustersAndTopics,
filteredNewClusterSplitStateMap);
// only restart if there was metadata change to handle duplicate
MetadataUpdateEvent from
// enumerator. We can possibly only restart the readers whose metadata
has changed but that
@@ -349,19 +361,11 @@ public class DynamicKafkaSourceReader<T> implements
SourceReader<T, DynamicKafka
// update event arrives. Splits in state could be
old and it's possible
// to not have another metadata update event, so
need to filter the
// splits at this point.
- .filter(
- pendingSplit -> {
- boolean splitValid =
- isSplitForActiveClusters(
- pendingSplit,
newClustersAndTopics);
- if (!splitValid) {
-
releaseOrDeferSplitOutput(pendingSplit.splitId());
- logger.info(
- "Removing invalid split
for reader: {}",
- pendingSplit);
- }
- return splitValid;
- })
+ .map(
+ pendingSplit ->
+ validatePendingSplit(
+ pendingSplit,
newClustersAndTopics))
+ .filter(split -> split != null)
.collect(Collectors.toList());
addSplits(validPendingSplits);
@@ -400,10 +404,12 @@ public class DynamicKafkaSourceReader<T> implements
SourceReader<T, DynamicKafka
@Override
public List<DynamicKafkaSourceSplit> snapshotState(long checkpointId) {
+ pruneExpiredRetainedSplits();
List<DynamicKafkaSourceSplit> splits =
snapshotStateFromAllReaders(checkpointId);
// pending splits should be typically empty, since we do not add
splits to pending splits if
// reader has started
+ splits.addAll(retainedSplits);
splits.addAll(pendingSplits);
return splits;
}
@@ -488,6 +494,7 @@ public class DynamicKafkaSourceReader<T> implements
SourceReader<T, DynamicKafka
kafkaClusterId,
clustersProperties.keySet()),
readerSpecificProperties);
+
DynamicKafkaSourceOptions.removeRemovedClusterRetentionOption(readerSpecificProperties);
KafkaPropertiesUtil.setClientIdPrefix(readerSpecificProperties,
kafkaClusterId);
// layer a kafka cluster group to distinguish metrics by cluster
@@ -625,6 +632,54 @@ public class DynamicKafkaSourceReader<T> implements
SourceReader<T, DynamicKafka
clustersProperties.clear();
}
+ private void reactivateRetainedSplits(
+ Map<String, Set<String>> newClustersAndTopics,
+ Map<String, List<KafkaPartitionSplit>>
filteredNewClusterSplitStateMap) {
+ List<DynamicKafkaSourceSplit> stillRetainedSplits = new ArrayList<>();
+ for (DynamicKafkaSourceSplit retainedSplit : retainedSplits) {
+ if (isSplitForActiveClusters(retainedSplit, newClustersAndTopics))
{
+ filteredNewClusterSplitStateMap
+ .computeIfAbsent(
+ retainedSplit.getKafkaClusterId(), (ignore) ->
new ArrayList<>())
+ .add(retainedSplit.clearRetention());
+ } else {
+ stillRetainedSplits.add(retainedSplit);
+ }
+ }
+
+ retainedSplits.clear();
+ retainedSplits.addAll(stillRetainedSplits);
+ }
+
+ private DynamicKafkaSourceSplit validatePendingSplit(
+ DynamicKafkaSourceSplit pendingSplit, Map<String, Set<String>>
newClustersAndTopics) {
+ if (isSplitForActiveClusters(pendingSplit, newClustersAndTopics)) {
+ return pendingSplit.clearRetention();
+ }
+
+ releaseOrDeferSplitOutput(pendingSplit.splitId());
+ if (pendingSplit.isRetained(System.currentTimeMillis())) {
+ retainedSplits.add(pendingSplit);
+ } else {
+ logger.info("Removing invalid split for reader: {}", pendingSplit);
+ }
+
+ return null;
+ }
+
+ private boolean shouldRetainSplit(
+ DynamicKafkaSourceSplit split, Map<String, Set<String>>
newClustersAndTopics) {
+ return removedClusterStateRetentionMs > 0
+ &&
!newClustersAndTopics.containsKey(split.getKafkaClusterId());
+ }
+
+ private void pruneExpiredRetainedSplits() {
+ long currentTimeMillis = System.currentTimeMillis();
+ retainedSplits.removeIf(
+ split -> split.isRetained() &&
!split.isRetained(currentTimeMillis));
+ pendingSplits.removeIf(split -> split.isRetained() &&
!split.isRetained(currentTimeMillis));
+ }
+
static Configuration toConfiguration(Properties props) {
Configuration config = new Configuration();
props.stringPropertyNames().forEach(key -> config.setString(key,
props.getProperty(key)));
diff --git
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/split/DynamicKafkaSourceSplit.java
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/split/DynamicKafkaSourceSplit.java
index b98cc805..85da9736 100644
---
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/split/DynamicKafkaSourceSplit.java
+++
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/split/DynamicKafkaSourceSplit.java
@@ -21,6 +21,8 @@ package org.apache.flink.connector.kafka.dynamic.source.split;
import org.apache.flink.annotation.Internal;
import org.apache.flink.connector.kafka.source.split.KafkaPartitionSplit;
+import javax.annotation.Nullable;
+
import java.util.Objects;
/** Split that wraps {@link KafkaPartitionSplit} with Kafka cluster
information. */
@@ -29,14 +31,23 @@ public class DynamicKafkaSourceSplit extends
KafkaPartitionSplit {
private final String kafkaClusterId;
private final KafkaPartitionSplit kafkaPartitionSplit;
+ @Nullable private final Long retainedUntilMs;
public DynamicKafkaSourceSplit(String kafkaClusterId, KafkaPartitionSplit
kafkaPartitionSplit) {
+ this(kafkaClusterId, kafkaPartitionSplit, null);
+ }
+
+ public DynamicKafkaSourceSplit(
+ String kafkaClusterId,
+ KafkaPartitionSplit kafkaPartitionSplit,
+ @Nullable Long retainedUntilMs) {
super(
kafkaPartitionSplit.getTopicPartition(),
kafkaPartitionSplit.getStartingOffset(),
kafkaPartitionSplit.getStoppingOffset().orElse(NO_STOPPING_OFFSET));
this.kafkaClusterId = kafkaClusterId;
this.kafkaPartitionSplit = kafkaPartitionSplit;
+ this.retainedUntilMs = retainedUntilMs;
}
@Override
@@ -52,6 +63,27 @@ public class DynamicKafkaSourceSplit extends
KafkaPartitionSplit {
return kafkaPartitionSplit;
}
+ @Nullable
+ public Long getRetainedUntilMs() {
+ return retainedUntilMs;
+ }
+
+ public boolean isRetained() {
+ return retainedUntilMs != null;
+ }
+
+ public boolean isRetained(long currentTimeMillis) {
+ return retainedUntilMs != null && retainedUntilMs > currentTimeMillis;
+ }
+
+ public DynamicKafkaSourceSplit retainUntil(long newRetainedUntilMs) {
+ return new DynamicKafkaSourceSplit(kafkaClusterId,
kafkaPartitionSplit, newRetainedUntilMs);
+ }
+
+ public DynamicKafkaSourceSplit clearRetention() {
+ return new DynamicKafkaSourceSplit(kafkaClusterId,
kafkaPartitionSplit);
+ }
+
@Override
public String toString() {
return "DynamicKafkaSourceSplit{"
@@ -60,6 +92,8 @@ public class DynamicKafkaSourceSplit extends
KafkaPartitionSplit {
+ '\''
+ ", kafkaPartitionSplit="
+ kafkaPartitionSplit
+ + ", retainedUntilMs="
+ + retainedUntilMs
+ '}';
}
@@ -76,11 +110,12 @@ public class DynamicKafkaSourceSplit extends
KafkaPartitionSplit {
}
DynamicKafkaSourceSplit that = (DynamicKafkaSourceSplit) o;
return Objects.equals(kafkaClusterId, that.kafkaClusterId)
- && Objects.equals(kafkaPartitionSplit,
that.kafkaPartitionSplit);
+ && Objects.equals(kafkaPartitionSplit,
that.kafkaPartitionSplit)
+ && Objects.equals(retainedUntilMs, that.retainedUntilMs);
}
@Override
public int hashCode() {
- return Objects.hash(super.hashCode(), kafkaClusterId,
kafkaPartitionSplit);
+ return Objects.hash(super.hashCode(), kafkaClusterId,
kafkaPartitionSplit, retainedUntilMs);
}
}
diff --git
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/split/DynamicKafkaSourceSplitSerializer.java
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/split/DynamicKafkaSourceSplitSerializer.java
index 69648374..9c7ed7d4 100644
---
a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/split/DynamicKafkaSourceSplitSerializer.java
+++
b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/split/DynamicKafkaSourceSplitSerializer.java
@@ -35,6 +35,7 @@ public class DynamicKafkaSourceSplitSerializer
implements SimpleVersionedSerializer<DynamicKafkaSourceSplit> {
private static final int VERSION_1 = 1;
+ private static final int VERSION_2 = 2;
private final KafkaPartitionSplitSerializer kafkaPartitionSplitSerializer;
@@ -44,7 +45,7 @@ public class DynamicKafkaSourceSplitSerializer
@Override
public int getVersion() {
- return VERSION_1;
+ return VERSION_2;
}
@Override
@@ -52,6 +53,11 @@ public class DynamicKafkaSourceSplitSerializer
try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputStream out = new DataOutputStream(baos)) {
out.writeUTF(split.getKafkaClusterId());
+ Long retainedUntilMs = split.getRetainedUntilMs();
+ out.writeBoolean(retainedUntilMs != null);
+ if (retainedUntilMs != null) {
+ out.writeLong(retainedUntilMs);
+ }
out.writeInt(kafkaPartitionSplitSerializer.getVersion());
out.write(kafkaPartitionSplitSerializer.serialize(split.getKafkaPartitionSplit()));
out.flush();
@@ -61,14 +67,27 @@ public class DynamicKafkaSourceSplitSerializer
@Override
public DynamicKafkaSourceSplit deserialize(int version, byte[] serialized)
throws IOException {
+ if (version != VERSION_1 && version != VERSION_2) {
+ throw new IOException(
+ String.format(
+ "The bytes are serialized with version %d, "
+ + "while this deserializer only supports
version up to %d",
+ version, getVersion()));
+ }
+
try (ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
DataInputStream in = new DataInputStream(bais)) {
String kafkaClusterId = in.readUTF();
+ Long retainedUntilMs = null;
+ if (version == VERSION_2 && in.readBoolean()) {
+ retainedUntilMs = in.readLong();
+ }
int kafkaPartitionSplitSerializerVersion = in.readInt();
KafkaPartitionSplit kafkaPartitionSplit =
kafkaPartitionSplitSerializer.deserialize(
kafkaPartitionSplitSerializerVersion,
in.readAllBytes());
- return new DynamicKafkaSourceSplit(kafkaClusterId,
kafkaPartitionSplit);
+ return new DynamicKafkaSourceSplit(
+ kafkaClusterId, kafkaPartitionSplit, retainedUntilMs);
}
}
}
diff --git
a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaTableFactory.java
b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaTableFactory.java
index 99304838..5514533f 100644
---
a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaTableFactory.java
+++
b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaTableFactory.java
@@ -122,6 +122,7 @@ public class DynamicKafkaTableFactory implements
DynamicTableSourceFactory {
options.add(SCAN_PARALLELISM);
options.add(DynamicKafkaSourceOptions.STREAM_METADATA_DISCOVERY_INTERVAL_MS);
options.add(DynamicKafkaSourceOptions.STREAM_METADATA_DISCOVERY_FAILURE_THRESHOLD);
+
options.add(DynamicKafkaSourceOptions.STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS);
options.add(DynamicKafkaSourceOptions.STREAM_ENUMERATOR_MODE);
return options;
}
@@ -217,6 +218,15 @@ public class DynamicKafkaTableFactory implements
DynamicTableSourceFactory {
.STREAM_METADATA_DISCOVERY_FAILURE_THRESHOLD
.key(),
Integer.toString(value)));
+ tableOptions
+
.getOptional(DynamicKafkaSourceOptions.STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS)
+ .ifPresent(
+ value ->
+ properties.setProperty(
+ DynamicKafkaSourceOptions
+
.STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS
+ .key(),
+ Long.toString(value)));
tableOptions
.getOptional(DynamicKafkaSourceOptions.STREAM_ENUMERATOR_MODE)
.ifPresent(
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/DynamicKafkaSourceITTest.java
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/DynamicKafkaSourceITTest.java
index c95f2f9a..a28e4412 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/DynamicKafkaSourceITTest.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/DynamicKafkaSourceITTest.java
@@ -18,14 +18,18 @@
package org.apache.flink.connector.kafka.dynamic.source;
+import org.apache.flink.api.common.JobStatus;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.connector.source.ReaderInfo;
import org.apache.flink.api.connector.source.SplitEnumerator;
import org.apache.flink.api.connector.source.SplitsAssignment;
import org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext;
+import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ExternalizedCheckpointRetention;
import org.apache.flink.configuration.RestartStrategyOptions;
+import org.apache.flink.configuration.StateBackendOptions;
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;
@@ -52,15 +56,21 @@ import
org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem
import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
import org.apache.flink.connector.testframe.testsuites.SourceTestSuiteBase;
import org.apache.flink.core.execution.CheckpointingMode;
+import org.apache.flink.core.execution.JobClient;
import org.apache.flink.core.testutils.CommonTestUtils;
import org.apache.flink.metrics.MetricGroup;
+import
org.apache.flink.runtime.messages.FlinkJobTerminatedWithoutCancellationException;
import org.apache.flink.runtime.testutils.InMemoryReporter;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.legacy.RichSinkFunction;
import
org.apache.flink.streaming.connectors.kafka.DynamicKafkaSourceTestHelper;
import org.apache.flink.streaming.connectors.kafka.KafkaTestBase;
import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.test.util.TestUtils;
+import org.apache.flink.testutils.junit.SharedObjectsExtension;
+import org.apache.flink.testutils.junit.SharedReference;
import org.apache.flink.util.CloseableIterator;
import com.google.common.collect.ImmutableList;
@@ -76,6 +86,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.extension.RegisterExtension;
import org.junit.jupiter.api.io.TempDir;
import java.io.File;
@@ -92,12 +103,16 @@ import java.util.NoSuchElementException;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
+import static
org.apache.flink.configuration.StateRecoveryOptions.SAVEPOINT_PATH;
import static
org.apache.flink.connector.kafka.dynamic.source.metrics.KafkaClusterMetricGroup.DYNAMIC_KAFKA_SOURCE_METRIC_GROUP;
import static org.assertj.core.api.Assertions.assertThat;
@@ -120,6 +135,9 @@ class DynamicKafkaSourceITTest {
@Nested
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
class DynamicKafkaSourceSpecificTests {
+ @RegisterExtension
+ final SharedObjectsExtension sharedObjects =
SharedObjectsExtension.create();
+
@BeforeAll
void beforeAll() throws Throwable {
DynamicKafkaSourceTestHelper.setup();
@@ -714,6 +732,97 @@ class DynamicKafkaSourceITTest {
.collect(Collectors.toList()));
}
+ @Test
+ void
testRemovedClusterOffsetsRetainedAcrossCheckpointRestoreAndRescale() throws
Throwable {
+ int kafkaClusterIdx = 0;
+ String topic = "test-retained-removed-cluster-" +
System.currentTimeMillis();
+ DynamicKafkaSourceTestHelper.createTopic(kafkaClusterIdx, topic,
NUM_PARTITIONS);
+
+ String testStreamId = "test-retained-removed-cluster-stream";
+ File metadataFile = File.createTempFile(testDir.getPath() +
"/metadata", ".yaml");
+ writeClusterMetadataToFile(
+ metadataFile,
+ testStreamId,
+ topic,
+ ImmutableList.of(
+
DynamicKafkaSourceTestHelper.getKafkaClusterTestEnvMetadata(
+ kafkaClusterIdx)));
+
+ SharedReference<List<Integer>> collectedRecords =
sharedObjects.add(new ArrayList<>());
+ Configuration checkpointConfiguration =
createCheckpointConfiguration();
+ JobClient phase1JobClient = null;
+ JobClient phase2JobClient = null;
+ try {
+ int stage1End =
+ DynamicKafkaSourceTestHelper.produceToKafka(
+ kafkaClusterIdx, topic, NUM_PARTITIONS,
NUM_RECORDS_PER_SPLIT, 0);
+ phase1JobClient =
+ startRetainedRemovedClusterJob(
+ checkpointConfiguration,
+ metadataFile,
+ testStreamId,
+ collectedRecords,
+ 1);
+ waitForCollectedRecords(
+ collectedRecords,
+ phase1JobClient,
+ stage1End,
+ "Could not read initial retained-cluster records
before removal");
+ assertThat(copyCollectedRecords(collectedRecords))
+ .containsExactlyInAnyOrderElementsOf(
+ IntStream.range(0,
stage1End).boxed().collect(Collectors.toList()));
+
+ File checkpointBeforeRemoval =
waitForCompletedCheckpoint(null);
+ writeClusterMetadataToFile(
+ metadataFile, testStreamId, topic,
Collections.emptyList());
+ waitForKafkaClusterMetricsToDisappear(
+ kafkaClusterTestEnvMetadata0.getKafkaClusterId());
+ waitForCompletedCheckpoint(checkpointBeforeRemoval);
+
+ phase1JobClient.cancel().get(30, TimeUnit.SECONDS);
+ phase1JobClient = null;
+ // The selected checkpoint can be subsumed before cancellation
completes.
+ File retainedCheckpoint = waitForCompletedCheckpoint(null);
+
+ writeClusterMetadataToFile(
+ metadataFile,
+ testStreamId,
+ topic,
+ ImmutableList.of(
+
DynamicKafkaSourceTestHelper.getKafkaClusterTestEnvMetadata(
+ kafkaClusterIdx)));
+ int stage2End =
+ DynamicKafkaSourceTestHelper.produceToKafka(
+ kafkaClusterIdx,
+ topic,
+ NUM_PARTITIONS,
+ NUM_RECORDS_PER_SPLIT,
+ stage1End);
+
+ Configuration restoreConfiguration = new
Configuration(checkpointConfiguration);
+ restoreConfiguration.set(SAVEPOINT_PATH,
retainedCheckpoint.toURI().toString());
+ phase2JobClient =
+ startRetainedRemovedClusterJob(
+ restoreConfiguration,
+ metadataFile,
+ testStreamId,
+ collectedRecords,
+ 2);
+ waitForCollectedRecords(
+ collectedRecords,
+ phase2JobClient,
+ stage2End,
+ "Could not read records after retained cluster re-add
and restore");
+
+ assertThat(copyCollectedRecords(collectedRecords))
+ .containsExactlyInAnyOrderElementsOf(
+ IntStream.range(0,
stage2End).boxed().collect(Collectors.toList()));
+ } finally {
+ cancelJob(phase2JobClient);
+ cancelJob(phase1JobClient);
+ }
+ }
+
@Test
void testTopicReAddMigrationUsingFileMetadataService() throws
Throwable {
// setup topics
@@ -1134,6 +1243,162 @@ class DynamicKafkaSourceITTest {
.collect(Collectors.toSet());
}
+ private Configuration createCheckpointConfiguration() {
+ Configuration configuration = new Configuration();
+ configuration.set(RestartStrategyOptions.RESTART_STRATEGY,
"disable");
+ configuration.set(StateBackendOptions.STATE_BACKEND, "rocksdb");
+ File checkpointDir = new File(testDir,
"retained-removed-cluster-checkpoints");
+ configuration.set(
+ CheckpointingOptions.CHECKPOINTS_DIRECTORY,
checkpointDir.toURI().toString());
+ configuration.set(
+ CheckpointingOptions.EXTERNALIZED_CHECKPOINT_RETENTION,
+ ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION);
+ configuration.set(CheckpointingOptions.MAX_RETAINED_CHECKPOINTS,
2);
+ return configuration;
+ }
+
+ private JobClient startRetainedRemovedClusterJob(
+ Configuration configuration,
+ File metadataFile,
+ String streamId,
+ SharedReference<List<Integer>> collectedRecords,
+ int parallelism)
+ throws Exception {
+ StreamExecutionEnvironment env =
+
StreamExecutionEnvironment.getExecutionEnvironment(configuration);
+ env.setParallelism(parallelism);
+ env.enableCheckpointing(100L);
+
+ Properties properties = new Properties();
+
properties.setProperty(KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS.key(),
"100");
+ properties.setProperty(
+
DynamicKafkaSourceOptions.STREAM_METADATA_DISCOVERY_INTERVAL_MS.key(), "100");
+ properties.setProperty(
+
DynamicKafkaSourceOptions.STREAM_METADATA_DISCOVERY_FAILURE_THRESHOLD.key(),
+ "2");
+ properties.setProperty(
+
DynamicKafkaSourceOptions.STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS.key(),
+ "60000");
+ properties.setProperty(
+ CommonClientConfigs.GROUP_ID_CONFIG,
"test-retained-removed-cluster-offsets");
+
+ YamlFileMetadataService yamlFileMetadataService =
+ new YamlFileMetadataService(metadataFile.getPath(),
Duration.ofMillis(100));
+ DynamicKafkaSource<Integer> dynamicKafkaSource =
+ DynamicKafkaSource.<Integer>builder()
+ .setStreamIds(Collections.singleton(streamId))
+ .setKafkaMetadataService(yamlFileMetadataService)
+ .setDeserializer(
+ KafkaRecordDeserializationSchema.valueOnly(
+ IntegerDeserializer.class))
+ .setStartingOffsets(OffsetsInitializer.earliest())
+ .setProperties(properties)
+ .build();
+
+ DataStreamSource<Integer> stream =
+ env.fromSource(
+ dynamicKafkaSource,
+ WatermarkStrategy.noWatermarks(),
+ "dynamic-kafka-src");
+ stream.uid("dynamic-kafka-src");
+ stream.addSink(new
CollectingSink(collectedRecords)).uid("collecting-sink");
+ return env.executeAsync("test-retained-removed-cluster-offsets");
+ }
+
+ private void waitForCollectedRecords(
+ SharedReference<List<Integer>> collectedRecords,
+ JobClient jobClient,
+ int expectedCount,
+ String message)
+ throws Exception {
+ CommonTestUtils.waitUtil(
+ () -> {
+ try {
+ throwIfJobFailed(jobClient);
+ } catch (Exception exception) {
+ throw new RuntimeException(exception);
+ }
+ return collectedRecords.applySync(
+ records -> records.size() >= expectedCount);
+ },
+ Duration.ofSeconds(30),
+ message);
+ }
+
+ private List<Integer> copyCollectedRecords(
+ SharedReference<List<Integer>> collectedRecords) {
+ return collectedRecords.applySync(ArrayList::new);
+ }
+
+ private void waitForKafkaClusterMetricsToDisappear(String
kafkaClusterId) throws Exception {
+ CommonTestUtils.waitUtil(
+ () -> !hasKafkaClusterMetrics(kafkaClusterId),
+ Duration.ofSeconds(30),
+ "Could not observe removed Kafka cluster metrics
disappear");
+ }
+
+ private boolean hasKafkaClusterMetrics(String kafkaClusterId) {
+ Optional<MetricGroup> groups =
reporter.findGroup(DYNAMIC_KAFKA_SOURCE_METRIC_GROUP);
+ if (groups.isEmpty()) {
+ return false;
+ }
+
+ return reporter.getMetricsByGroup(groups.get()).keySet().stream()
+ .map(metricName ->
groups.get().getMetricIdentifier(metricName))
+ .anyMatch(metricName ->
metricName.contains(".kafkaCluster." + kafkaClusterId));
+ }
+
+ private File waitForCompletedCheckpoint(File previousCheckpoint)
throws Exception {
+ AtomicReference<File> completedCheckpoint = new
AtomicReference<>();
+ CommonTestUtils.waitUtil(
+ () -> {
+ try {
+ File checkpoint =
+ TestUtils.getMostRecentCompletedCheckpoint(
+ new File(
+ testDir,
+
"retained-removed-cluster-checkpoints"));
+ if (checkpoint != null &&
!checkpoint.equals(previousCheckpoint)) {
+ completedCheckpoint.set(checkpoint);
+ return true;
+ }
+ } catch (Exception ignored) {
+ // Checkpoint directory is not populated yet.
+ }
+ return false;
+ },
+ Duration.ofSeconds(30),
+ "Could not obtain a completed retained-cluster
checkpoint");
+ return completedCheckpoint.get();
+ }
+
+ private void cancelJob(JobClient jobClient) throws Exception {
+ if (jobClient != null) {
+ try {
+ jobClient.cancel().get(30, TimeUnit.SECONDS);
+ } catch (ExecutionException executionException) {
+ if (!(executionException.getCause()
+ instanceof
FlinkJobTerminatedWithoutCancellationException)) {
+ throw executionException;
+ }
+ }
+ }
+ }
+
+ private void throwIfJobFailed(JobClient jobClient) throws Exception {
+ if (jobClient.getJobStatus().get(30, TimeUnit.SECONDS) !=
JobStatus.FAILED) {
+ return;
+ }
+
+ try {
+ jobClient.getJobExecutionResult().get(30, TimeUnit.SECONDS);
+ } catch (ExecutionException executionException) {
+ throw new RuntimeException(
+ "Dynamic source job failed before expected records
were collected",
+ executionException.getCause());
+ }
+ }
+
private void registerReader(
MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context,
DynamicKafkaSourceEnumerator enumerator,
@@ -1207,6 +1472,19 @@ class DynamicKafkaSourceITTest {
}
}
+ private static final class CollectingSink extends
RichSinkFunction<Integer> {
+ private final SharedReference<List<Integer>> collectedRecords;
+
+ private CollectingSink(SharedReference<List<Integer>>
collectedRecords) {
+ this.collectedRecords = collectedRecords;
+ }
+
+ @Override
+ public void invoke(Integer value, Context context) {
+ collectedRecords.consumeSync(records -> records.add(value));
+ }
+ }
+
/** Integration test based on connector testing framework. */
@Nested
class IntegrationTests extends SourceTestSuiteBase<String> {
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 1473bbdc..d79e9323 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
@@ -104,7 +104,15 @@ public class DynamicKafkaSourceEnumStateSerializerTest {
getSplitAssignment("topic3",
1, UNASSIGNED),
getSplitAssignment("topic4",
2, UNASSIGNED),
getSplitAssignment("topic5",
3, UNASSIGNED)),
- false)));
+ false)),
+ ImmutableMap.of(
+ "retained-cluster",
+ new
DynamicKafkaSourceEnumState.RetainedClusterState(
+ new KafkaSourceEnumState(
+ ImmutableSet.of(
+
getSplitAssignment("topic6", 0, ASSIGNED)),
+ true),
+ 123L)));
DynamicKafkaSourceEnumState dynamicKafkaSourceEnumStateAfterSerde =
dynamicKafkaSourceEnumStateSerializer.deserialize(
@@ -190,6 +198,7 @@ public class DynamicKafkaSourceEnumStateSerializerTest {
dynamicKafkaSourceEnumStateSerializer.deserialize(1,
serializedState);
assertThat(dynamicKafkaSourceEnumState.getClusterEnumeratorStates()).isEmpty();
+
assertThat(dynamicKafkaSourceEnumState.getRetainedClusterEnumeratorStates()).isEmpty();
KafkaStream kafkaStream =
dynamicKafkaSourceEnumState.getKafkaStreams().iterator().next();
assertThat(kafkaStream.getStreamId()).isEqualTo("stream0");
ClusterMetadata clusterMetadata =
kafkaStream.getClusterMetadataMap().get("cluster0");
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 e6867d32..4683037d 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
@@ -681,6 +681,163 @@ public class DynamicKafkaSourceEnumeratorTest {
}
}
+ @Test
+ public void testEnumeratorStateRetainsRemovedClusterUntilExpired() throws
Throwable {
+ int restoredParallelism = NUM_SUBTASKS + 1;
+ KafkaStream initialStream =
DynamicKafkaSourceTestHelper.getKafkaStream(TOPIC);
+ KafkaStream shrunkStream =
DynamicKafkaSourceTestHelper.getKafkaStream(TOPIC);
+ String removedClusterId =
DynamicKafkaSourceTestHelper.getKafkaClusterId(1);
+ shrunkStream.getClusterMetadataMap().remove(removedClusterId);
+
+ try (MockKafkaMetadataService metadataService =
+ new
MockKafkaMetadataService(Collections.singleton(initialStream));
+ MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS);
+ DynamicKafkaSourceEnumerator enumerator =
+ createEnumerator(
+ context,
+ metadataService,
+ (properties) -> {
+ properties.setProperty(
+ DynamicKafkaSourceOptions
+
.STREAM_METADATA_DISCOVERY_INTERVAL_MS
+ .key(),
+ "1");
+ properties.setProperty(
+ DynamicKafkaSourceOptions
+
.STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS
+ .key(),
+ "1000");
+ })) {
+ enumerator.start();
+ context.runPeriodicCallable(0);
+ runAllOneTimeCallables(context);
+
+ mockRegisterReaderAndSendReaderStartupEvent(context, enumerator,
0);
+ mockRegisterReaderAndSendReaderStartupEvent(context, enumerator,
1);
+ runAllOneTimeCallables(context);
+
+
metadataService.setKafkaStreams(Collections.singleton(shrunkStream));
+ context.runPeriodicCallable(0);
+ runAllOneTimeCallables(context);
+
+ DynamicKafkaSourceEnumState retainedCheckpoint =
enumerator.snapshotState(-1);
+ assertThat(retainedCheckpoint.getClusterEnumeratorStates())
+ .doesNotContainKey(removedClusterId);
+ assertThat(retainedCheckpoint.getRetainedClusterEnumeratorStates())
+ .containsKey(removedClusterId);
+
+ Properties restoredProperties = new Properties();
+ restoredProperties.setProperty(
+ KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS.key(),
"0");
+ restoredProperties.setProperty(
+
DynamicKafkaSourceOptions.STREAM_METADATA_DISCOVERY_INTERVAL_MS.key(), "1");
+ restoredProperties.setProperty(
+
DynamicKafkaSourceOptions.STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS.key(),
+ "1000");
+ try (MockKafkaMetadataService restoredMetadataService =
+ new
MockKafkaMetadataService(Collections.singleton(shrunkStream));
+ MockSplitEnumeratorContext<DynamicKafkaSourceSplit>
restoredContext =
+ new
MockSplitEnumeratorContext<>(restoredParallelism);
+ DynamicKafkaSourceEnumerator restoredEnumerator =
+ new DynamicKafkaSourceEnumerator(
+ new
KafkaStreamSetSubscriber(Collections.singleton(TOPIC)),
+ restoredMetadataService,
+ restoredContext,
+ OffsetsInitializer.earliest(),
+ new NoStoppingOffsetsInitializer(),
+ restoredProperties,
+ Boundedness.CONTINUOUS_UNBOUNDED,
+ retainedCheckpoint,
+ new TestKafkaEnumContextProxyFactory())) {
+ restoredEnumerator.start();
+ restoredContext.runPeriodicCallable(0);
+ runAllOneTimeCallables(restoredContext);
+ for (int i = 0; i < restoredParallelism; i++) {
+ mockRegisterReaderAndSendReaderStartupEvent(
+ restoredContext, restoredEnumerator, i);
+ }
+ runAllOneTimeCallables(restoredContext);
+
+ int assignmentsBeforeReAdd =
restoredContext.getSplitsAssignmentSequence().size();
+
restoredMetadataService.setKafkaStreams(Collections.singleton(initialStream));
+ restoredContext.runPeriodicCallable(0);
+ runAllOneTimeCallables(restoredContext);
+ assertThat(restoredContext.getSplitsAssignmentSequence())
+ .as(
+ "re-added cluster should not resend splits
restored from retained state after rescale")
+ .hasSize(assignmentsBeforeReAdd);
+ }
+
+ DynamicKafkaSourceEnumState expiredCheckpoint = retainedCheckpoint;
+ DynamicKafkaSourceEnumState.RetainedClusterState
retainedClusterState =
+
expiredCheckpoint.getRetainedClusterEnumeratorStates().get(removedClusterId);
+ expiredCheckpoint
+ .getRetainedClusterEnumeratorStates()
+ .put(
+ removedClusterId,
+ new
DynamicKafkaSourceEnumState.RetainedClusterState(
+
retainedClusterState.getKafkaSourceEnumState(),
+ System.currentTimeMillis() - 1));
+
+ Properties expiredStateProperties = new Properties();
+ expiredStateProperties.setProperty(
+
DynamicKafkaSourceOptions.STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS.key(),
+ "1000");
+ try (MockKafkaMetadataService restoredMetadataService =
+ new
MockKafkaMetadataService(Collections.singleton(shrunkStream));
+ MockSplitEnumeratorContext<DynamicKafkaSourceSplit>
restoredContext =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS);
+ DynamicKafkaSourceEnumerator restoredEnumerator =
+ new DynamicKafkaSourceEnumerator(
+ new
KafkaStreamSetSubscriber(Collections.singleton(TOPIC)),
+ restoredMetadataService,
+ restoredContext,
+ OffsetsInitializer.earliest(),
+ new NoStoppingOffsetsInitializer(),
+ expiredStateProperties,
+ Boundedness.CONTINUOUS_UNBOUNDED,
+ expiredCheckpoint,
+ new TestKafkaEnumContextProxyFactory())) {
+ assertThat(
+ restoredEnumerator
+ .snapshotState(-1)
+ .getRetainedClusterEnumeratorStates())
+ .doesNotContainKey(removedClusterId);
+ }
+ }
+ }
+
+ @Test
+ public void
testRemovedClusterRetentionOptionIsNotPassedToClusterEnumerator() throws
Exception {
+ MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context =
+ new MockSplitEnumeratorContext<>(2);
+ try (DynamicKafkaSourceEnumerator enumerator =
+ createEnumerator(
+ context,
+ properties ->
+ properties.setProperty(
+ DynamicKafkaSourceOptions
+
.STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS
+ .key(),
+ "60000"))) {
+ enumerator.start();
+ mockRegisterReaderAndSendReaderStartupEvent(context, enumerator,
0);
+
+ Map<String, ?> clusterEnumeratorMap =
+ (Map<String, ?>) Whitebox.getInternalState(enumerator,
"clusterEnumeratorMap");
+ for (Object clusterEnumerator : clusterEnumeratorMap.values()) {
+ Properties clusterEnumeratorProperties =
+ (Properties)
Whitebox.getInternalState(clusterEnumerator, "properties");
+ assertThat(clusterEnumeratorProperties)
+ .doesNotContainKey(
+ DynamicKafkaSourceOptions
+
.STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS
+ .key());
+ }
+ }
+ }
+
@Test
public void testStartupWithCheckpointState() throws Throwable {
// init enumerator with checkpoint state
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderTest.java
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderTest.java
index f7c91fb5..c8e95ef3 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderTest.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderTest.java
@@ -27,6 +27,7 @@ import
org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
import org.apache.flink.connector.base.source.reader.splitreader.SplitsChange;
import
org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
import org.apache.flink.connector.kafka.dynamic.metadata.KafkaStream;
+import
org.apache.flink.connector.kafka.dynamic.source.DynamicKafkaSourceOptions;
import org.apache.flink.connector.kafka.dynamic.source.MetadataUpdateEvent;
import
org.apache.flink.connector.kafka.dynamic.source.split.DynamicKafkaSourceSplit;
import
org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics;
@@ -191,6 +192,59 @@ public class DynamicKafkaSourceReaderTest extends
SourceReaderTestBase<DynamicKa
}
}
+ @Test
+ void testHandleSourceEventRetainsRemovedClusterOffsetsUntilExpired()
throws Exception {
+ DynamicKafkaSourceSplit cluster0Split =
+ new DynamicKafkaSourceSplit(
+ kafkaClusterId0, new KafkaPartitionSplit(new
TopicPartition(TOPIC, 0), 10));
+ DynamicKafkaSourceSplit cluster1Split =
+ new DynamicKafkaSourceSplit(
+ kafkaClusterId1, new KafkaPartitionSplit(new
TopicPartition(TOPIC, 0), 20));
+ KafkaStream shrunkKafkaStream =
DynamicKafkaSourceTestHelper.getKafkaStream(TOPIC);
+ shrunkKafkaStream.getClusterMetadataMap().remove(kafkaClusterId0);
+
+ TestingReaderContext context = new TestingReaderContext();
+ try (DynamicKafkaSourceReader<Integer> reader =
+ createReaderWithoutStartWithRemovedClusterRetention(context)) {
+ reader.start();
+ KafkaStream kafkaStream =
DynamicKafkaSourceTestHelper.getKafkaStream(TOPIC);
+ reader.handleSourceEvents(new
MetadataUpdateEvent(Collections.singleton(kafkaStream)));
+
+ reader.addSplits(ImmutableList.of(cluster0Split, cluster1Split));
+
+ reader.handleSourceEvents(
+ new
MetadataUpdateEvent(Collections.singleton(shrunkKafkaStream)));
+
+ DynamicKafkaSourceSplit retainedCluster0Split =
+ reader.snapshotState(-1).stream()
+ .filter(split ->
split.getKafkaClusterId().equals(kafkaClusterId0))
+ .findFirst()
+ .orElseThrow();
+ assertThat(retainedCluster0Split.isRetained()).isTrue();
+ assertThat(retainedCluster0Split.getRetainedUntilMs())
+ .isGreaterThan(System.currentTimeMillis());
+ assertThat(reader.snapshotState(-1))
+ .containsExactlyInAnyOrder(retainedCluster0Split,
cluster1Split);
+
+ reader.handleSourceEvents(new
MetadataUpdateEvent(Collections.singleton(kafkaStream)));
+ assertThat(reader.snapshotState(-1))
+ .containsExactlyInAnyOrder(cluster0Split, cluster1Split);
+ }
+
+ TestingReaderContext restoredContext = new TestingReaderContext();
+ try (DynamicKafkaSourceReader<Integer> restoredReader =
+
createReaderWithoutStartWithRemovedClusterRetention(restoredContext)) {
+ restoredReader.addSplits(
+ ImmutableList.of(
+
cluster0Split.retainUntil(System.currentTimeMillis() - 1),
+ cluster1Split));
+ restoredReader.start();
+ restoredReader.handleSourceEvents(
+ new
MetadataUpdateEvent(Collections.singleton(shrunkKafkaStream)));
+
assertThat(restoredReader.snapshotState(-1)).containsExactly(cluster1Split);
+ }
+ }
+
@Test
void testNoSubReadersInputStatus() throws Exception {
try (DynamicKafkaSourceReader<Integer> reader =
@@ -345,6 +399,18 @@ public class DynamicKafkaSourceReaderTest extends
SourceReaderTestBase<DynamicKa
properties);
}
+ private DynamicKafkaSourceReader<Integer>
createReaderWithoutStartWithRemovedClusterRetention(
+ TestingReaderContext context) {
+ Properties properties = getRequiredProperties();
+ properties.setProperty(
+
DynamicKafkaSourceOptions.STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS.key(),
+ "1000");
+ return new DynamicKafkaSourceReader<>(
+ context,
+
KafkaRecordDeserializationSchema.valueOnly(IntegerDeserializer.class),
+ properties);
+ }
+
private SourceReader<Integer, DynamicKafkaSourceSplit> startReader(
DynamicKafkaSourceReader<Integer> reader, TestingReaderContext
context) {
reader.start();
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/split/DynamicKafkaSourceSplitSerializerTest.java
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/split/DynamicKafkaSourceSplitSerializerTest.java
index 4125219a..850bbf20 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/split/DynamicKafkaSourceSplitSerializerTest.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/split/DynamicKafkaSourceSplitSerializerTest.java
@@ -19,10 +19,13 @@
package org.apache.flink.connector.kafka.dynamic.source.split;
import org.apache.flink.connector.kafka.source.split.KafkaPartitionSplit;
+import
org.apache.flink.connector.kafka.source.split.KafkaPartitionSplitSerializer;
import org.apache.kafka.common.TopicPartition;
import org.junit.jupiter.api.Test;
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
import java.io.IOException;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -41,7 +44,51 @@ public class DynamicKafkaSourceSplitSerializerTest {
"test-cluster",
new KafkaPartitionSplit(new
TopicPartition("test-topic", 3), 1));
DynamicKafkaSourceSplit dynamicKafkaSourceSplitAfterSerde =
- serializer.deserialize(1,
serializer.serialize(dynamicKafkaSourceSplit));
+ serializer.deserialize(
+ serializer.getVersion(),
serializer.serialize(dynamicKafkaSourceSplit));
assertEquals(dynamicKafkaSourceSplit,
dynamicKafkaSourceSplitAfterSerde);
}
+
+ @Test
+ public void testSerdeRetainedSplit() throws IOException {
+ DynamicKafkaSourceSplitSerializer serializer = new
DynamicKafkaSourceSplitSerializer();
+ DynamicKafkaSourceSplit retainedSplit =
+ new DynamicKafkaSourceSplit(
+ "test-cluster",
+ new KafkaPartitionSplit(new
TopicPartition("test-topic", 3), 1),
+ 123L);
+
+ DynamicKafkaSourceSplit retainedSplitAfterSerde =
+ serializer.deserialize(
+ serializer.getVersion(),
serializer.serialize(retainedSplit));
+
+ assertEquals(retainedSplit, retainedSplitAfterSerde);
+ }
+
+ @Test
+ public void testDeserializeV1State() throws IOException {
+ DynamicKafkaSourceSplitSerializer serializer = new
DynamicKafkaSourceSplitSerializer();
+ DynamicKafkaSourceSplit dynamicKafkaSourceSplit =
+ serializer.deserialize(1, serializeV1State());
+
+ assertEquals(
+ new DynamicKafkaSourceSplit(
+ "test-cluster",
+ new KafkaPartitionSplit(new
TopicPartition("test-topic", 3), 1)),
+ dynamicKafkaSourceSplit);
+ }
+
+ private static byte[] serializeV1State() throws IOException {
+ KafkaPartitionSplitSerializer kafkaPartitionSplitSerializer =
+ new KafkaPartitionSplitSerializer();
+ KafkaPartitionSplit kafkaPartitionSplit =
+ new KafkaPartitionSplit(new TopicPartition("test-topic", 3),
1);
+ try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(baos)) {
+ out.writeUTF("test-cluster");
+ out.writeInt(kafkaPartitionSplitSerializer.getVersion());
+
out.write(kafkaPartitionSplitSerializer.serialize(kafkaPartitionSplit));
+ return baos.toByteArray();
+ }
+ }
}
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaTableFactoryTest.java
b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaTableFactoryTest.java
index db7b4715..3bb5deae 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaTableFactoryTest.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaTableFactoryTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.streaming.connectors.kafka.table;
+import
org.apache.flink.connector.kafka.dynamic.source.DynamicKafkaSourceOptions;
import org.apache.flink.streaming.connectors.kafka.config.BoundedMode;
import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
import org.apache.flink.table.api.DataTypes;
@@ -77,6 +78,25 @@ class DynamicKafkaTableFactoryTest {
.withMessageContaining("stream-pattern");
}
+ @Test
+ void testTableSourceWithRemovedClusterRetentionOption() {
+ final Map<String, String> options = getSingleClusterSourceOptions();
+ options.put(
+
DynamicKafkaSourceOptions.STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS.key(),
+ "60000");
+
+ final DynamicTableSource actualSource = createTableSource(SCHEMA,
options);
+ assertThat(actualSource).isInstanceOf(DynamicKafkaTableSource.class);
+
+ final DynamicKafkaTableSource actualKafkaSource =
(DynamicKafkaTableSource) actualSource;
+ assertThat(
+ actualKafkaSource.properties.getProperty(
+ DynamicKafkaSourceOptions
+
.STREAM_METADATA_REMOVED_CLUSTER_RETENTION_MS
+ .key()))
+ .isEqualTo("60000");
+ }
+
private static Map<String, String> getSingleClusterSourceOptions() {
Map<String, String> tableOptions = new HashMap<>();
// Dynamic Kafka specific options.
