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:36 UTC
(flink-connector-kafka) 04/04: [FLINK-40128][connector] Topic integrity ITCase
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 db222b3c1e56e737b64b904e16a01b393aa2d932
Author: Efrat Levitan <el...@confluent.io>
AuthorDate: Sun Jul 12 20:25:16 2026 +0300
[FLINK-40128][connector] Topic integrity ITCase
---
.../kafka/source/SourceTopicIntegrityTest.java | 280 +++++++++++++++++++++
.../kafka/testutils/KafkaSourceTestEnv.java | 71 +++++-
2 files changed, 341 insertions(+), 10 deletions(-)
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/SourceTopicIntegrityTest.java b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/SourceTopicIntegrityTest.java
new file mode 100644
index 00000000..43a05b7b
--- /dev/null
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/SourceTopicIntegrityTest.java
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.kafka.source;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.StateRecoveryOptions;
+import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
+import org.apache.flink.connector.kafka.sink.KafkaSink;
+import org.apache.flink.connector.kafka.source.enumerator.metadata.TopicIntegrityException;
+import org.apache.flink.connector.kafka.testutils.KafkaSourceTestEnv;
+import org.apache.flink.core.execution.SavepointFormatType;
+import org.apache.flink.core.testutils.CommonTestUtils;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.minicluster.MiniCluster;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
+import org.apache.flink.test.junit5.InjectMiniCluster;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+
+import org.apache.kafka.common.TopicPartition;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.api.parallel.ResourceLock;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.Properties;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+/** Integration tests for topic integrity checking in KafkaSource. */
+@ResourceLock("KafkaTestBase")
+public class SourceTopicIntegrityTest {
+ private static final Logger LOG = LoggerFactory.getLogger(SourceTopicIntegrityTest.class);
+ private static final String SOURCE_TOPIC_NAME = "SourceTopicIntegrityTest_source-topic";
+ private static final String SOURCE_TOPIC_PATTERN = "SourceTopicIntegrityTest_source.*";
+ private static final String SINK_TOPIC_NAME = "SourceTopicIntegrityTest_sink-topic";
+ private static final long DISCOVERY_INTERVAL = 50L;
+ private static final Duration ERROR_DISCOVERY_TIMEOUT = Duration.ofSeconds(2);
+ @TempDir private Path savepointBasePath;
+
+ @RegisterExtension
+ public static final MiniClusterExtension MINI_CLUSTER_RESOURCE =
+ new MiniClusterExtension(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(3)
+ .setNumberSlotsPerTaskManager(3)
+ .build());
+
+ public void setupKafka(boolean createTopics) throws Throwable {
+ KafkaSourceTestEnv.setup();
+ if (createTopics) {
+ KafkaSourceTestEnv.createTestTopic(SOURCE_TOPIC_NAME);
+ KafkaSourceTestEnv.produceToKafka(
+ KafkaSourceTestEnv.getRecordsForTopic(SOURCE_TOPIC_NAME));
+ KafkaSourceTestEnv.createTestTopic(SINK_TOPIC_NAME);
+ }
+ }
+
+ @BeforeEach
+ public void setup() throws Throwable {
+ setupKafka(true);
+ }
+
+ @AfterEach
+ public void tearDown() throws Exception {
+ KafkaSourceTestEnv.tearDown();
+ }
+
+ private static Properties getSourceProperties() {
+ final Properties props = new Properties();
+ props.setProperty(
+ KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS.key(),
+ String.valueOf(DISCOVERY_INTERVAL));
+ return props;
+ }
+
+ private final JobGraph getJobGraph(
+ Configuration extraConf, SourceSubscriptionMode sourceSubscriptionMode)
+ throws Throwable {
+ KafkaSourceBuilder<String> sourceBuilder =
+ KafkaSource.<String>builder()
+ .setBootstrapServers(KafkaSourceTestEnv.brokerConnectionStrings)
+ .setValueOnlyDeserializer(new SimpleStringSchema())
+ .enableTopicIntegrityCheck()
+ .setProperties(getSourceProperties());
+ switch (sourceSubscriptionMode) {
+ case PARTITIONS:
+ sourceBuilder.setPartitions(
+ IntStream.range(0, KafkaSourceTestEnv.NUM_PARTITIONS)
+ .mapToObj(i -> new TopicPartition(SOURCE_TOPIC_NAME, i))
+ .collect(Collectors.toSet()));
+ break;
+ case TOPICS:
+ sourceBuilder.setTopics(SOURCE_TOPIC_NAME);
+ break;
+ case PATTERN:
+ sourceBuilder.setTopicPattern(Pattern.compile(SOURCE_TOPIC_PATTERN));
+ break;
+ default:
+ throw new IllegalStateException(
+ "Unsupported subscription mode " + sourceSubscriptionMode);
+ }
+
+ KafkaSource<String> source = sourceBuilder.build();
+ KafkaSink<String> sink =
+ KafkaSink.<String>builder()
+ .setBootstrapServers(KafkaSourceTestEnv.brokerConnectionStrings)
+ .setRecordSerializer(
+ KafkaRecordSerializationSchema.builder()
+ .setTopic(SINK_TOPIC_NAME)
+ .setValueSerializationSchema(new SimpleStringSchema())
+ .setPartitioner(new FlinkFixedPartitioner())
+ .build())
+ .build();
+ Configuration configuration = new Configuration();
+ configuration.addAll(extraConf);
+ final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment(configuration);
+ DataStream<String> stream =
+ env.fromSource(source, WatermarkStrategy.noWatermarks(), "kafka-source");
+ stream.sinkTo(sink);
+ return env.getStreamGraph().getJobGraph();
+ }
+
+ private static Stream<Arguments> testTopicIntegrityFailureArgsProvider() {
+ return Stream.of(
+ Arguments.of(SourceSubscriptionMode.PARTITIONS, false),
+ Arguments.of(SourceSubscriptionMode.TOPICS, false),
+ Arguments.of(SourceSubscriptionMode.PATTERN, false),
+ Arguments.of(SourceSubscriptionMode.PARTITIONS, true),
+ Arguments.of(SourceSubscriptionMode.TOPICS, true),
+ Arguments.of(SourceSubscriptionMode.PATTERN, true));
+ }
+
+ /**
+ * Test that job fails if the source topic id does not match the current topic id.
+ *
+ * @throws Throwable
+ */
+ @ParameterizedTest
+ @MethodSource("testTopicIntegrityFailureArgsProvider")
+ public void testTopicIntegrityFailure(
+ SourceSubscriptionMode sourceSubscriptionMode,
+ boolean recreateTopic,
+ @InjectMiniCluster MiniCluster miniCluster)
+ throws Throwable {
+ JobGraph firstJobGraph = getJobGraph(new Configuration(), sourceSubscriptionMode);
+ miniCluster.submitJob(firstJobGraph).get();
+ org.apache.flink.runtime.testutils.CommonTestUtils.waitForAllTaskRunning(
+ miniCluster, firstJobGraph.getJobID(), true);
+ String savepointPath =
+ miniCluster
+ .stopWithSavepoint(
+ firstJobGraph.getJobID(),
+ savepointBasePath.toFile().toString(),
+ false,
+ SavepointFormatType.DEFAULT)
+ .get();
+
+ // Restart kafka to simulate topic recreation
+ tearDown();
+ setupKafka(recreateTopic);
+
+ // resume from savepoint
+ Configuration configuration = new Configuration();
+ configuration.set(StateRecoveryOptions.SAVEPOINT_PATH, savepointPath);
+ JobGraph secondJobGraph = getJobGraph(configuration, sourceSubscriptionMode);
+ miniCluster.submitJob(secondJobGraph).get();
+ final JobID secondJobId = secondJobGraph.getJobID();
+ final String expectedError =
+ String.format(
+ recreateTopic ? "%s: Topic %s was recreated" : "%s: Topic %s is missing",
+ TopicIntegrityException.class.getName(),
+ SOURCE_TOPIC_NAME);
+ CommonTestUtils.waitUtil(
+ () -> {
+ try {
+ return miniCluster
+ .getArchivedExecutionGraph(secondJobId)
+ .get()
+ .getFailureInfo()
+ .getExceptionAsString()
+ .contains(expectedError);
+ } catch (Exception e) {
+ LOG.warn(
+ "Error while fetching execution graph for job {}: {}",
+ secondJobId,
+ e.getMessage());
+ return false;
+ }
+ },
+ ERROR_DISCOVERY_TIMEOUT,
+ Duration.ofMillis(DISCOVERY_INTERVAL),
+ "Waiting for job to fail with " + expectedError);
+ }
+
+ /**
+ * Test that job can resume and run successfully if the source topic id matches the one before
+ * restart.
+ *
+ * @throws Throwable
+ */
+ @ParameterizedTest
+ @EnumSource(
+ value = SourceSubscriptionMode.class,
+ names = {"PARTITIONS", "TOPICS", "PATTERN"})
+ public void testTopicIntegritySuccess(
+ SourceSubscriptionMode sourceSubscriptionMode,
+ @InjectMiniCluster MiniCluster miniCluster)
+ throws Throwable {
+ JobGraph firstJobGraph = getJobGraph(new Configuration(), sourceSubscriptionMode);
+ miniCluster.submitJob(firstJobGraph).get();
+ org.apache.flink.runtime.testutils.CommonTestUtils.waitForAllTaskRunning(
+ miniCluster, firstJobGraph.getJobID(), true);
+ final int initialExpectedRecords =
+ KafkaSourceTestEnv.NUM_RECORDS_PER_PARTITION * KafkaSourceTestEnv.NUM_PARTITIONS;
+ KafkaSourceTestEnv.waitForRecordsInTopic(SINK_TOPIC_NAME, initialExpectedRecords);
+ String savepointPath =
+ miniCluster
+ .stopWithSavepoint(
+ firstJobGraph.getJobID(),
+ savepointBasePath.toFile().toString(),
+ false,
+ SavepointFormatType.DEFAULT)
+ .get();
+
+ // generate more records to the same topic
+ KafkaSourceTestEnv.produceToKafka(KafkaSourceTestEnv.getRecordsForTopic(SOURCE_TOPIC_NAME));
+ final int expectedTotalRecords = 2 * initialExpectedRecords;
+
+ // resume from savepoint
+ Configuration configuration = new Configuration();
+ configuration.set(StateRecoveryOptions.SAVEPOINT_PATH, savepointPath);
+ JobGraph secondJobGraph = getJobGraph(configuration, sourceSubscriptionMode);
+ miniCluster.submitJob(secondJobGraph).get();
+ final JobID secondJobId = secondJobGraph.getJobID();
+ org.apache.flink.runtime.testutils.CommonTestUtils.waitForAllTaskRunning(
+ miniCluster, secondJobId, true);
+
+ // Expect the job has run and produced the extra records to the sink
+ KafkaSourceTestEnv.waitForRecordsInTopic(SINK_TOPIC_NAME, expectedTotalRecords);
+ }
+
+ private enum SourceSubscriptionMode {
+ PARTITIONS,
+ TOPICS,
+ PATTERN
+ }
+}
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/KafkaSourceTestEnv.java b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/KafkaSourceTestEnv.java
index ed4a10ef..715fb483 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/KafkaSourceTestEnv.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/testutils/KafkaSourceTestEnv.java
@@ -19,21 +19,24 @@
package org.apache.flink.connector.kafka.testutils;
import org.apache.flink.connector.kafka.source.split.KafkaPartitionSplit;
+import org.apache.flink.core.testutils.CommonTestUtils;
import org.apache.flink.streaming.connectors.kafka.KafkaTestBase;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsSpec;
import org.apache.kafka.clients.admin.RecordsToDelete;
+import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.IntegerSerializer;
-import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -53,6 +56,9 @@ public class KafkaSourceTestEnv extends KafkaTestBase {
public static final int NUM_PARTITIONS = 10;
public static final int NUM_RECORDS_PER_PARTITION = 10;
+ private static final Duration WAIT_DATA_TIMEOUT = Duration.ofMinutes(2);
+ private static final Duration WAIT_DATA_INTERVAL = Duration.ofSeconds(2);
+
private static AdminClient adminClient;
private static KafkaConsumer<String, Integer> consumer;
@@ -77,15 +83,7 @@ public class KafkaSourceTestEnv extends KafkaTestBase {
}
public static KafkaConsumer<String, Integer> getConsumer() {
- Properties props = new Properties();
- props.putAll(standardProps);
- props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
- props.setProperty(
- ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
- props.setProperty(
- ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
- IntegerDeserializer.class.getName());
- return new KafkaConsumer<>(props);
+ return new KafkaConsumer<>(getConsumerProperties(IntegerDeserializer.class));
}
public static Properties getConsumerProperties(Class<?> deserializerClass) {
@@ -274,4 +272,57 @@ public class KafkaSourceTestEnv extends KafkaTestBase {
setupCommittedOffsets(topic);
}
}
+
+ public static String getTopicId(String topicName)
+ throws InterruptedException, ExecutionException {
+ return getAdminClient()
+ .listTopics()
+ .namesToListings()
+ .get()
+ .get(topicName)
+ .topicId()
+ .toString();
+ }
+
+ private static long getRecordCountInTopic(String topicName) {
+ try (Consumer<String, Integer> consumer = getConsumer()) {
+ List<TopicPartition> partitions =
+ consumer.partitionsFor(topicName).stream()
+ .map(info -> new TopicPartition(topicName, info.partition()))
+ .collect(Collectors.toList());
+
+ consumer.assign(partitions);
+ consumer.seekToBeginning(partitions);
+
+ long totalRecords = 0;
+ ConsumerRecords<String, Integer> records;
+ do {
+ records = consumer.poll(Duration.ofSeconds(1));
+ totalRecords += records.count();
+ } while (!records.isEmpty());
+
+ return totalRecords;
+ }
+ }
+
+ public static void waitForRecordsInTopic(String topicName, int expectedCount) throws Exception {
+ CommonTestUtils.waitUtil(
+ () -> {
+ long count = KafkaSourceTestEnv.getRecordCountInTopic(topicName);
+ LOG.info(
+ "Found {} records in topic {} (expected: {})",
+ count,
+ topicName,
+ expectedCount);
+ if (count == expectedCount) {
+ return true;
+ }
+ return false;
+ },
+ WAIT_DATA_TIMEOUT,
+ WAIT_DATA_INTERVAL,
+ String.format(
+ "Timeout waiting for %d records in topic %s after %s",
+ expectedCount, topicName, WAIT_DATA_TIMEOUT));
+ }
}