You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fa...@apache.org on 2026/09/03 09:31:34 UTC
(flink-connector-kafka) 02/04: [FLINK-40128][connector] Add an opt-in based topic integrity config option
This is an automated email from the ASF dual-hosted git repository.
1996fanrui pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
commit 9cc364d80db5c3906dd4afe2b8817b90c548fb36
Author: Efrat Levitan <el...@confluent.io>
AuthorDate: Sun Jul 12 20:22:56 2026 +0300
[FLINK-40128][connector] Add an opt-in based topic integrity config option
Currently, when a user decides to delete/recreate a kafka source topic, the flink job silently adapts. This commit adds an opt-in config allowing to verify the integrity of a kafka source topic and fail it if it is missing in broker invenroty (deleted) or presented with a different topic id (=recreated)
---
.../content.zh/docs/connectors/datastream/kafka.md | 32 +++++++++++++++++++
docs/content/docs/connectors/datastream/kafka.md | 37 ++++++++++++++++++++++
.../connector/kafka/source/KafkaSourceBuilder.java | 26 +++++++++++++++
.../connector/kafka/source/KafkaSourceOptions.java | 7 ++++
4 files changed, 102 insertions(+)
diff --git a/docs/content.zh/docs/connectors/datastream/kafka.md b/docs/content.zh/docs/connectors/datastream/kafka.md
index 26c1f2ac..52b2691e 100644
--- a/docs/content.zh/docs/connectors/datastream/kafka.md
+++ b/docs/content.zh/docs/connectors/datastream/kafka.md
@@ -220,6 +220,8 @@ Kafka Source 支持流式和批式两种运行模式。默认情况下，KafkaSo
- ```commit.offsets.on.checkpoint``` 指定是否在进行 checkpoint 时将消费位点提交至 Kafka broker
- ```poll.timeout.ms``` 指定 Kafka Consumer 单次 poll 等待数据的最长时间（毫秒），默认为 10 秒。
减小该值可以让空闲的 Source 更快地响应分片变更，但 poll 会更频繁
+- ```scan.topic-integrity-check.enabled``` 指定是否在运行时校验订阅 Topic 的 id，并在 Topic 缺失或被重建时使作业失败。
+ 默认关闭。详见下面的<a href="#topic-integrity-check">Topic 完整性检查</a>一节
Kafka consumer 的配置可以参考 [Apache Kafka 文档](http://kafka.apache.org/documentation/#consumerconfigs)。
@@ -251,6 +253,36 @@ KafkaSource.builder() \
分区检查间隔默认为5分钟。需要显式地设置分区检查间隔为非正数才能关闭此功能。
{{< /hint >}}
+### Topic 完整性检查
+如果某个订阅的 Topic 被删除后又以相同的名称重建，Kafka Source 默认会像什么都没发生一样，继续从新的 Topic 静默地消费数据，
+而这通常并非用户所期望的行为。Topic 完整性检查通过跟踪每个订阅 Topic 的 topic id 来防止这种情况：一旦检测到 topic id
+不匹配（Topic 被重建）或 Topic 缺失，就会使作业失败，而不是静默地继续从不同的底层 Topic 中读取数据。
+
+要启用该功能，可以在 ```KafkaSourceBuilder``` 上调用 ```enableTopicIntegrityCheck()```，或者将属性
+```scan.topic-integrity-check.enabled``` 设置为 ```true```：
+
+{{< tabs "KafkaSource#TopicIntegrityCheck" >}}
+{{< tab "Java" >}}
+```java
+KafkaSource.builder()
+ .enableTopicIntegrityCheck();
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+{{< hint info >}}
+Topic 完整性检查仅在通过 ```setTopics(String...)```、```setTopicPattern(Pattern)``` 或
+```setPartitions(Set)``` 订阅 Topic 时受支持。如果将此选项与未实现 ```TopicMetadataSettable``` 接口的
+自定义 ```KafkaSubscriber```（通过 ```setKafkaSubscriber``` 设置）一起启用，构建器会在构建时抛出异常。
+{{< /hint >}}
+
+该检查会在 Source 每次从 Kafka 拉取 Topic / Partition 元数据时执行，即启动时执行一次；如果启用了
+<a href="#dynamic-partition-discovery">动态分区检查</a>，之后每个检查周期还会再执行一次。要在作业运行期间
+周期性地执行该检查，请确保 ```partition.discovery.interval.ms``` 被设置为正值；否则该检查只会在
+Source 启动或恢复时执行一次。
+
+当某个 Topic 被检测为缺失或被重建时，Source 会抛出 ```TopicIntegrityException``` 并使作业失败。
+
### 事件时间和水印
默认情况下，Kafka Source 使用 Kafka 消息中的时间戳作为事件时间。您可以定义自己的水印策略（Watermark Strategy）
以从消息中提取事件时间，并向下游发送水印：
diff --git a/docs/content/docs/connectors/datastream/kafka.md b/docs/content/docs/connectors/datastream/kafka.md
index 20290297..d134a17b 100644
--- a/docs/content/docs/connectors/datastream/kafka.md
+++ b/docs/content/docs/connectors/datastream/kafka.md
@@ -231,6 +231,9 @@ metric group
- ```poll.timeout.ms``` defines the maximum time in milliseconds the Kafka consumer blocks in a
single poll while waiting for records, 10 seconds by default. Lowering it makes an idle source
react faster to split changes, at the cost of polling more often
+- ```scan.topic-integrity-check.enabled``` specifies whether to verify the id of the subscribed
+ topics during runtime and fail the job if a topic is missing or was recreated. Disabled by
+ default. See <a href="#topic-integrity-check">Topic Integrity Check</a> below for more details.
For configurations of KafkaConsumer, you can refer to
<a href="http://kafka.apache.org/documentation/#consumerconfigs">Apache Kafka documentation</a>
@@ -269,6 +272,40 @@ KafkaSource.builder() \
The partition discovery interval is 5 minutes by default. To **disable** this feature, you need to explicitly set the partition discovery interval to a non-positive value.
{{< /hint >}}
+### Topic Integrity Check
+If a subscribed topic is deleted and later recreated with the same name, the Kafka source would by
+default silently keep consuming from the new topic as if nothing happened, which is often not what
+users want. Topic integrity check protects against this scenario by tracking the topic id of every
+subscribed topic and failing the job as soon as a mismatch (topic recreated) or a missing topic is
+detected, instead of silently continuing to read from a different underlying topic.
+
+To enable this feature, either call ```enableTopicIntegrityCheck()``` on the ```KafkaSourceBuilder```,
+or set the property ```scan.topic-integrity-check.enabled``` to ```true```:
+
+{{< tabs "KafkaSource#TopicIntegrityCheck" >}}
+{{< tab "Java" >}}
+```java
+KafkaSource.builder()
+ .enableTopicIntegrityCheck();
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+{{< hint info >}}
+Topic integrity check is only supported when the source subscribes to topics using
+```setTopics(String...)```, ```setTopicPattern(Pattern)``` or ```setPartitions(Set)```. Enabling
+this option together with a custom ```KafkaSubscriber``` (via ```setKafkaSubscriber```) that does
+not implement ```TopicMetadataSettable``` will cause the builder to throw an exception at build time.
+{{< /hint >}}
+
+The check is performed every time the source fetches topic/partition metadata from Kafka, i.e. once
+at startup and, if <a href="#dynamic-partition-discovery">partition discovery</a> is enabled, once
+per discovery interval afterwards. To have the check performed periodically while the job is
+running, make sure ```partition.discovery.interval.ms``` is set to a positive value; otherwise the
+topics are only checked once during source startup or recovery.
+
+When a topic is found missing or recreated, the source fails with a ```TopicIntegrityException```.
+
### Event Time and Watermarks
By default, the record will use the timestamp embedded in Kafka ```ConsumerRecord``` as the event
time. You can define your own ```WatermarkStrategy``` for extract event time from the record itself,
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java
index 4167f385..b88d6711 100644
--- a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java
+++ b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java
@@ -24,6 +24,7 @@ import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.connector.kafka.source.enumerator.initializer.NoStoppingOffsetsInitializer;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializerValidator;
+import org.apache.flink.connector.kafka.source.enumerator.metadata.TopicMetadataSettable;
import org.apache.flink.connector.kafka.source.enumerator.subscriber.KafkaSubscriber;
import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
import org.apache.flink.util.function.SerializableSupplier;
@@ -375,6 +376,19 @@ public class KafkaSourceBuilder<OUT> {
return this;
}
+ /**
+ * Whether perform a check for the integrity of source topics, and fail the job if the topic was
+ * deleted or recreated. For the check to be performed periodically during source runtime,
+ * partition discovery must be enabled (KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS set
+ * to a positive number)
+ *
+ * @return {@link KafkaSourceBuilder}
+ */
+ public KafkaSourceBuilder<OUT> enableTopicIntegrityCheck() {
+ this.setProperty(KafkaSourceOptions.TOPIC_INTEGRITY_CHECK_ENABLED.key(), "true");
+ return this;
+ }
+
/**
* Set an arbitrary property for the KafkaSource and KafkaConsumer. The valid keys can be found
* in {@link ConsumerConfig} and {@link KafkaSourceOptions}.
@@ -561,6 +575,12 @@ public class KafkaSourceBuilder<OUT> {
if (props.containsKey(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG)) {
checkDeserializer(props.getProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG));
}
+ if (topicIntegrityCheckEnabled()) {
+ checkState(
+ subscriber instanceof TopicMetadataSettable,
+ "Topic integrity check is not supported for non TopicMetadataSettable subscriber %s",
+ subscriber.getClass().getName());
+ }
}
private void checkDeserializer(String deserializer) {
@@ -610,4 +630,10 @@ public class KafkaSourceBuilder<OUT> {
KafkaSourceOptions.COMMIT_OFFSETS_ON_CHECKPOINT.key()));
return autoCommit || commitOnCheckpoint;
}
+
+ private boolean topicIntegrityCheckEnabled() {
+ return props.containsKey(KafkaSourceOptions.TOPIC_INTEGRITY_CHECK_ENABLED.key())
+ && Boolean.parseBoolean(
+ props.getProperty(KafkaSourceOptions.TOPIC_INTEGRITY_CHECK_ENABLED.key()));
+ }
}
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceOptions.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceOptions.java
index 5928d8a2..188c8f9d 100644
--- a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceOptions.java
+++ b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceOptions.java
@@ -70,6 +70,13 @@ public class KafkaSourceOptions {
+ "of polling more often. 0 polls without blocking. Must not be "
+ "negative.");
+ public static final ConfigOption<Boolean> TOPIC_INTEGRITY_CHECK_ENABLED =
+ ConfigOptions.key("scan.topic-integrity-check.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to verify topic id during runtime and fail if the topic is missing or recreated");
+
@SuppressWarnings("unchecked")
public static <T> T getOption(
Properties props, ConfigOption<?> configOption, Function<String, T> parser) {