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 564ad828e43e650a579fa17073156ebf4d129047 Author: Efrat Levitan <[email protected]> AuthorDate: Sun Jul 12 20:18:54 2026 +0300 [FLINK-40128][connector] Introduce topic integrity utilities in static kafka source Currently, when a user decides to delete/recreate a kafka source topic, the flink job silently adapts. This commits added the required utilities 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) If UNKNOWN_TOPIC_OR_PARTITION was thrown during topic discovery, topic integrity provider will issue a followup call (list topics) to determine if the exception was thrown due to topic missing in metadata, and if yes, will throw a TopicIntegrityException("topic <> is missing"). If no topic is missing, the original UNKNOWN_TOPIC_OR_PARTITION will be thrown for retry. (The reasoning for throwing TopicIntegrityException over UNKNOWN_TOPIC_OR_PARTITION is that TopicIntegrityException tri [...] --- flink-connector-kafka/pom.xml | 8 + .../metadata/TopicIntegrityException.java | 29 +++ .../metadata/TopicIntegrityProvider.java | 170 ++++++++++++++++++ .../enumerator/metadata/TopicMetadataProvider.java | 58 ++++++ .../enumerator/metadata/TopicMetadataSettable.java | 28 +++ .../enumerator/TopicIntegrityProviderTest.java | 195 +++++++++++++++++++++ 6 files changed, 488 insertions(+) diff --git a/flink-connector-kafka/pom.xml b/flink-connector-kafka/pom.xml index bf3addd5..088e950d 100644 --- a/flink-connector-kafka/pom.xml +++ b/flink-connector-kafka/pom.xml @@ -164,6 +164,14 @@ under the License. <!-- Tests --> + <dependency> + <groupId>org.apache.kafka</groupId> + <artifactId>kafka-clients</artifactId> + <version>${kafka.version}</version> + <classifier>test</classifier> + <scope>test</scope> + </dependency> + <dependency> <groupId>com.google.guava</groupId> <artifactId>guava</artifactId> diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicIntegrityException.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicIntegrityException.java new file mode 100644 index 00000000..2a56b088 --- /dev/null +++ b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicIntegrityException.java @@ -0,0 +1,29 @@ +/* + * 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.enumerator.metadata; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.util.FlinkRuntimeException; + +/** Exception thrown when topic integrity check fails. */ +@Internal +public class TopicIntegrityException extends FlinkRuntimeException { + public TopicIntegrityException(String message) { + super(message); + } +} diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicIntegrityProvider.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicIntegrityProvider.java new file mode 100644 index 00000000..896f0ccd --- /dev/null +++ b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicIntegrityProvider.java @@ -0,0 +1,170 @@ +/* + * 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.enumerator.metadata; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.connector.kafka.util.AdminUtils; +import org.apache.flink.util.ExceptionUtils; + +import org.apache.kafka.clients.admin.AdminClient; +import org.apache.kafka.clients.admin.TopicDescription; +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.errors.UnknownTopicOrPartitionException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.regex.Pattern; +import java.util.stream.Collectors; + +/** + * Provider of topic integrity related functionalities for {@link + * org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumerator}. + */ +@Internal +public class TopicIntegrityProvider implements TopicMetadataProvider { + + private static final Logger LOG = LoggerFactory.getLogger(TopicIntegrityProvider.class); + private final Map<String, String> trackedTopicIdsByName; + + public TopicIntegrityProvider(Map<String, String> trackedTopicIdsByNameFromContext) { + trackedTopicIdsByName = new ConcurrentHashMap<>(trackedTopicIdsByNameFromContext); + } + + @Override + public Map<String, TopicDescription> getTopicMetadata( + AdminClient adminClient, Pattern pattern) { + final Collection<String> topicsToVerifyInPatternMode = + trackedTopicIdsByName.keySet().stream() + .filter(name -> pattern.matcher(name).matches()) + .collect(Collectors.toCollection(HashSet::new)); + topicsToVerifyInPatternMode.addAll(AdminUtils.getTopicsByPattern(adminClient, pattern)); + return getTopicMetadata(adminClient, topicsToVerifyInPatternMode); + } + + @Override + public Map<String, TopicDescription> getTopicMetadata( + AdminClient adminClient, Collection<String> subscribedTopicNames) { + Map<String, TopicDescription> topicMetadata; + try { + topicMetadata = AdminUtils.getTopicMetadata(adminClient, subscribedTopicNames); + failIfRecreated(subscribedTopicNames, topicMetadata); + } catch (RuntimeException original) { + if (ExceptionUtils.findThrowable(original, UnknownTopicOrPartitionException.class) + .isPresent()) { + // UnknownTopicOrPartitionException can be transient due to broker timeout + // or permanent due to topic/partition loss. + // Determine if the exception is caused by a missing topic + // and if yes, trigger a TopicIntegrity failure instead + try { + failIfMissing(subscribedTopicNames, adminClient.listTopics().names().get()); + } catch (TopicIntegrityException missingTopicException) { + throw missingTopicException; + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + } catch (Exception ignored) { + // ignored so we fallback to the original error + } + } + throw original; + } + refreshTrackedTopicIds(subscribedTopicNames, topicMetadata); + return topicMetadata; + } + + private void refreshTrackedTopicIds( + Collection<String> subscribedTopicNames, Map<String, TopicDescription> topicMetadata) { + + // Add new subscribed topic to trackedTopicIdsByName + for (String subscribedTopicName : subscribedTopicNames) { + if (!trackedTopicIdsByName.containsKey(subscribedTopicName)) { + final Uuid topicId = topicMetadata.get(subscribedTopicName).topicId(); + if (topicId == null || topicId.equals(Uuid.ZERO_UUID)) { + continue; + } + trackedTopicIdsByName.put(subscribedTopicName, topicId.toString()); + } + } + // Remove outdated topics from trackedTopicIdsByName + for (String topicNameFromMapping : trackedTopicIdsByName.keySet()) { + if (!subscribedTopicNames.contains(topicNameFromMapping)) { + trackedTopicIdsByName.remove(topicNameFromMapping); + } + } + } + + public Map<String, String> getTrackedTopicIdsByName() { + return new HashMap<>(trackedTopicIdsByName); + } + + private void failIfRecreated( + Collection<String> subscribedTopicNames, Map<String, TopicDescription> metadataTopics) + throws RuntimeException { + for (String subscribedTopicName : subscribedTopicNames) { + final TopicDescription topicDescription = metadataTopics.get(subscribedTopicName); + if (topicDescription == null) { + LOG.error("Topic {} found missing during recreation check", subscribedTopicName); + throw new TopicIntegrityException("Topic " + subscribedTopicName + " is missing"); + } + final String topicIdFromState = trackedTopicIdsByName.get(subscribedTopicName); + final Uuid topicIdFromMetadata = topicDescription.topicId(); + if (topicIdFromState == null + || topicIdFromMetadata == null + || topicIdFromMetadata.equals(Uuid.ZERO_UUID)) { + // we skip topic integrity check for null topicId + // due to broker configuration, or topic not yet stored on trackedTopicIdsByName + LOG.warn( + "Topic integrity check skipped due to a null topicId: topic name: {}," + + " topic id passed from initial config: {}" + + " current topic id on kafka server: {}", + subscribedTopicName, + topicIdFromState, + topicIdFromMetadata); + continue; + } + if (!topicIdFromState.equals(topicIdFromMetadata.toString())) { + LOG.error( + "Topic integrity mismatch: expected topic Id of {} to be {}, got {}", + subscribedTopicName, + topicIdFromState, + topicIdFromMetadata); + throw new TopicIntegrityException( + "Topic " + subscribedTopicName + " was recreated"); + } + } + } + + private static void failIfMissing( + Collection<String> subscribedTopicNames, Collection<String> existingTopicNames) + throws RuntimeException { + for (String subscribedTopicName : subscribedTopicNames) { + if (!existingTopicNames.contains(subscribedTopicName)) { + LOG.error( + "Topic {} is missing in current topics {}", + subscribedTopicName, + existingTopicNames); + throw new TopicIntegrityException("Topic " + subscribedTopicName + " is missing"); + } + } + } +} diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicMetadataProvider.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicMetadataProvider.java new file mode 100644 index 00000000..cf4a9dcf --- /dev/null +++ b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicMetadataProvider.java @@ -0,0 +1,58 @@ +/* + * 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.enumerator.metadata; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.connector.kafka.util.AdminUtils; + +import org.apache.kafka.clients.admin.AdminClient; +import org.apache.kafka.clients.admin.TopicDescription; + +import java.util.Collection; +import java.util.Map; +import java.util.regex.Pattern; + +/** + * Interface for providing topic integrity mapping to subscribers that are aware of topic integrity. + */ +@Internal +public interface TopicMetadataProvider { + + Map<String, TopicDescription> getTopicMetadata( + AdminClient adminClient, Collection<String> subscribedTopicNames); + + Map<String, TopicDescription> getTopicMetadata(AdminClient adminClient, Pattern pattern); + + static TopicMetadataProvider createDefault() { + return new TopicMetadataProvider() { + @Override + public Map<String, TopicDescription> getTopicMetadata( + AdminClient adminClient, Collection<String> subscribedTopicNames) + throws RuntimeException { + return AdminUtils.getTopicMetadata(adminClient, subscribedTopicNames); + } + + @Override + public Map<String, TopicDescription> getTopicMetadata( + AdminClient adminClient, Pattern pattern) throws RuntimeException { + return AdminUtils.getTopicMetadata(adminClient, pattern); + } + }; + } +} diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicMetadataSettable.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicMetadataSettable.java new file mode 100644 index 00000000..d193bc02 --- /dev/null +++ b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicMetadataSettable.java @@ -0,0 +1,28 @@ +/* + * 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.enumerator.metadata; + +import org.apache.flink.annotation.Internal; + +/** Interface for setting a custom {@link TopicMetadataProvider} other than the default. */ +@Internal +public interface TopicMetadataSettable { + + void setTopicMetadataProvider(TopicMetadataProvider topicMetadataProvider); +} diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/TopicIntegrityProviderTest.java b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/TopicIntegrityProviderTest.java new file mode 100644 index 00000000..bfd9d277 --- /dev/null +++ b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/TopicIntegrityProviderTest.java @@ -0,0 +1,195 @@ +/* + * 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.enumerator; + +import org.apache.flink.connector.kafka.source.enumerator.metadata.TopicIntegrityException; +import org.apache.flink.connector.kafka.source.enumerator.metadata.TopicIntegrityProvider; + +import org.apache.kafka.clients.admin.MockAdminClient; +import org.apache.kafka.clients.admin.TopicDescription; +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.errors.TimeoutException; +import org.apache.kafka.common.errors.UnknownTopicOrPartitionException; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.regex.Pattern; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.api.Assertions.entry; + +/** Unit tests for {@link TopicIntegrityProvider}. */ +class TopicIntegrityProviderTest { + + private static final String TOPIC1 = "topic1"; + private static final String TOPIC2 = "topic2"; + private static MockAdminClient mockAdmin; + + @BeforeEach + public void setup() { + mockAdmin = new MockAdminClient(); + } + + @Test + void testReturnsVerifiedTopics() throws Exception { + String id = addTopic(TOPIC1); + + TopicIntegrityProvider provider = new TopicIntegrityProvider(Map.of(TOPIC1, id)); + + Map<String, TopicDescription> result = + provider.getTopicMetadata(mockAdmin, Collections.singletonList(TOPIC1)); + + assertThat(result).containsOnlyKeys(TOPIC1); + assertThat(result.get(TOPIC1).topicId().toString()).isEqualTo(id); + } + + @Test + void testAddsNewlySubscribedTopicWithoutFailingIntegrityCheck() throws Exception { + String id = addTopic(TOPIC1); + + TopicIntegrityProvider provider = new TopicIntegrityProvider(new HashMap<>()); + + provider.getTopicMetadata(mockAdmin, Collections.singletonList(TOPIC1)); + + assertThat(provider.getTrackedTopicIdsByName()).containsExactly(entry(TOPIC1, id)); + } + + @Test + void testFailsIfTopicIsMissing() { + TopicIntegrityProvider provider = + new TopicIntegrityProvider(Map.of(TOPIC1, Uuid.randomUuid().toString())); + + assertThatThrownBy( + () -> + provider.getTopicMetadata( + mockAdmin, Collections.singletonList(TOPIC1))) + .isInstanceOf(TopicIntegrityException.class) + .hasMessageContaining("Topic " + TOPIC1 + " is missing"); + } + + @Test + void testFailsIfTopicWasRecreated() throws Exception { + String originalId = addTopic(TOPIC1); + + TopicIntegrityProvider provider = new TopicIntegrityProvider(Map.of(TOPIC1, originalId)); + + // Simulate recreation: delete and re-add under the same name, yielding a new id. + mockAdmin.deleteTopics(Collections.singletonList(TOPIC1)).all().get(); + addTopic(TOPIC1); + + assertThatThrownBy( + () -> + provider.getTopicMetadata( + mockAdmin, Collections.singletonList(TOPIC1))) + .isInstanceOf(TopicIntegrityException.class) + .hasMessageContaining("Topic " + TOPIC1 + " was recreated"); + } + + @Test + void testThrowsOriginalErrorWhenUnknownTopicExceptionIsNotDueToMissingTopic() throws Exception { + String id = addTopic(TOPIC1); + mockAdmin.markTopicForDeletion(TOPIC1); + + TopicIntegrityProvider provider = new TopicIntegrityProvider(Map.of(TOPIC1, id)); + + assertThatThrownBy( + () -> + provider.getTopicMetadata( + mockAdmin, Collections.singletonList(TOPIC1))) + .isNotInstanceOf(TopicIntegrityException.class) + .hasRootCauseInstanceOf(UnknownTopicOrPartitionException.class); + } + + @Test + void testThrowsOriginalErrorForUnrelatedException() throws Exception { + String id = addTopic(TOPIC1); + mockAdmin.timeoutNextRequest(1); + + TopicIntegrityProvider provider = new TopicIntegrityProvider(Map.of(TOPIC1, id)); + + assertThatThrownBy( + () -> + provider.getTopicMetadata( + mockAdmin, Collections.singletonList(TOPIC1))) + .isNotInstanceOf(TopicIntegrityException.class) + .hasRootCauseInstanceOf(TimeoutException.class); + } + + @Test + void testRemovesOutdatedTopicFromMapping() throws Exception { + String id1 = addTopic(TOPIC1); + + Map<String, String> tracked = new HashMap<>(); + tracked.put(TOPIC1, id1); + tracked.put(TOPIC2, Uuid.randomUuid().toString()); + TopicIntegrityProvider provider = new TopicIntegrityProvider(tracked); + + // Only TOPIC1 is subscribed to anymore; TOPIC2 must be dropped from the tracked mapping. + provider.getTopicMetadata(mockAdmin, Collections.singletonList(TOPIC1)); + + assertThat(provider.getTrackedTopicIdsByName()).containsExactly(entry(TOPIC1, id1)); + } + + @Test + void testPatternModeStillChecksTopicsThatDisappearedFromLiveMatch() { + TopicIntegrityProvider provider = + new TopicIntegrityProvider(Map.of(TOPIC1, Uuid.randomUuid().toString())); + + assertThatThrownBy(() -> provider.getTopicMetadata(mockAdmin, Pattern.compile(".*"))) + .isInstanceOf(TopicIntegrityException.class) + .hasMessageContaining("Topic " + TOPIC1 + " is missing"); + } + + @Test + void testGetTrackedTopicIdsByNameReturnsDefensiveCopy() { + TopicIntegrityProvider provider = + new TopicIntegrityProvider(Map.of(TOPIC1, Uuid.randomUuid().toString())); + + Map<String, String> mapping = provider.getTrackedTopicIdsByName(); + mapping.put(TOPIC2, Uuid.randomUuid().toString()); + + assertThat(provider.getTrackedTopicIdsByName()).doesNotContainKey(TOPIC2); + } + + @Test + void testEmptySubscriptionReturnsEmptyMetadataWithoutError() { + TopicIntegrityProvider provider = new TopicIntegrityProvider(new HashMap<>()); + + Map<String, TopicDescription> result = + provider.getTopicMetadata(mockAdmin, Collections.emptyList()); + + assertThat(result).isEmpty(); + assertThat(provider.getTrackedTopicIdsByName()).isEmpty(); + } + + private static String addTopic(String name) throws Exception { + mockAdmin.addTopic(false, name, Collections.emptyList(), Collections.emptyMap()); + return mockAdmin + .describeTopics(Collections.singletonList(name)) + .allTopicNames() + .get() + .get(name) + .topicId() + .toString(); + } +}
