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 <[email protected]> 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) {
