Savonitar commented on code in PR #287: URL: https://github.com/apache/flink-connector-kafka/pull/287#discussion_r3604032718
########## flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/subscriber/TopicIntegrityProvider.java: ########## @@ -0,0 +1,162 @@ +/* + * 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.subscriber; + +import org.apache.flink.connector.kafka.integrity.TopicIntegrityException; + +import org.apache.commons.lang3.exception.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.io.Serializable; +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; + +import static org.apache.flink.connector.kafka.util.AdminUtils.getTopicMetadata; +import static org.apache.flink.connector.kafka.util.AdminUtils.getTopicsByPattern; + +/** Provider of topic integrity related functionalities for {@link KafkaSourceEnumerator}. */ Review Comment: does this render properly on your side? I can't click on it in IDE (Cannot resolve the `KafkaSourceEnumerator` and it is red), meanwhile other java doc links resolve correctly. Maybe it can be fully qualified: ```suggestion /** Provider of topic integrity related functionalities for {@link org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumerator}. */ ``` Does it work properly on your side? ########## flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/subscriber/TopicIntegrityProvider.java: ########## @@ -0,0 +1,162 @@ +/* + * 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.subscriber; + +import org.apache.flink.connector.kafka.integrity.TopicIntegrityException; + +import org.apache.commons.lang3.exception.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.io.Serializable; +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; + +import static org.apache.flink.connector.kafka.util.AdminUtils.getTopicMetadata; +import static org.apache.flink.connector.kafka.util.AdminUtils.getTopicsByPattern; + +/** Provider of topic integrity related functionalities for {@link KafkaSourceEnumerator}. */ +class TopicIntegrityProvider implements Serializable { + + private static final Logger LOG = LoggerFactory.getLogger(TopicIntegrityProvider.class); + private final Map<String, String> topicIntegrityMapping = new ConcurrentHashMap<>(); + + TopicIntegrityProvider() {} + + public void open(Map<String, String> topicIntegrityMappingFromContext) { + topicIntegrityMapping.putAll(topicIntegrityMappingFromContext); + } + + public Map<String, TopicDescription> getVerifiedTopicMetadata( + AdminClient adminClient, Pattern pattern) { + final Collection<String> topicsToVerifyInPatternMode = + topicIntegrityMapping.keySet().stream() + .filter(pattern.asPredicate()) + .collect(Collectors.toCollection(HashSet::new)); + topicsToVerifyInPatternMode.addAll(getTopicsByPattern(adminClient, pattern)); + return getVerifiedTopicMetadata(adminClient, topicsToVerifyInPatternMode); + } + + public Map<String, TopicDescription> getVerifiedTopicMetadata( + AdminClient adminClient, Collection<String> subscribedTopicNames) { + Map<String, TopicDescription> topicMetadata = new HashMap<>(); + try { + topicMetadata = getTopicMetadata(adminClient, subscribedTopicNames); + failIfRecreated(subscribedTopicNames, topicMetadata); + } catch (RuntimeException original) { + if (ExceptionUtils.getRootCause(original) instanceof UnknownTopicOrPartitionException) { + // 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 (Exception ignored) { + // ignored so we fallback to the original error + } + } + throw original; + } + trackNewIdsInMapping(subscribedTopicNames, topicMetadata); + return topicMetadata; + } + + private void trackNewIdsInMapping( + Collection<String> subscribedTopicNames, Map<String, TopicDescription> topicMetadata) { + + // Add new subscribed topic to mapping + for (String subscribedTopicName : subscribedTopicNames) { + if (!topicIntegrityMapping.keySet().contains(subscribedTopicName)) { + topicIntegrityMapping.put( + subscribedTopicName, + topicMetadata.get(subscribedTopicName).topicId().toString()); + } + } + // Remove outdated topics from mapping + for (String topicNameFromMapping : topicIntegrityMapping.keySet()) { + if (!subscribedTopicNames.contains(topicNameFromMapping)) { + topicIntegrityMapping.remove(topicNameFromMapping); + } + } + } + + public Map<String, String> getTopicIntegrityMapping() { + return new HashMap<>(topicIntegrityMapping); + } + + 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 = topicIntegrityMapping.get(subscribedTopicName); + final Uuid topicIdFromMetadata = topicDescription.topicId(); + if (topicIdFromState == null || topicIdFromMetadata == null) { Review Comment: Can `topicIdFromMetadata` be null? From the Kafka client source code, it looks like it is always `Uuid.ZERO_UUID`. ########## flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/subscriber/TopicIntegrityProvider.java: ########## @@ -0,0 +1,162 @@ +/* + * 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.subscriber; + +import org.apache.flink.connector.kafka.integrity.TopicIntegrityException; + +import org.apache.commons.lang3.exception.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.io.Serializable; +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; + +import static org.apache.flink.connector.kafka.util.AdminUtils.getTopicMetadata; +import static org.apache.flink.connector.kafka.util.AdminUtils.getTopicsByPattern; + +/** Provider of topic integrity related functionalities for {@link KafkaSourceEnumerator}. */ +class TopicIntegrityProvider implements Serializable { + + private static final Logger LOG = LoggerFactory.getLogger(TopicIntegrityProvider.class); + private final Map<String, String> topicIntegrityMapping = new ConcurrentHashMap<>(); Review Comment: `topicIntegrityMapping` name does not seem to be informative. "Mapping" says nothing about what maps to what, or why. Many people follow Java convention for a Map field which is to encode the key->value relationship: `valuesByKey` or `keyToValue`. Maybe we can use `knownTopicIdsByName` or `trackedTopicIdsByName`, or you have a better naming. wdyt? ########## flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/subscriber/TopicIntegrityProvider.java: ########## @@ -0,0 +1,162 @@ +/* + * 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.subscriber; + +import org.apache.flink.connector.kafka.integrity.TopicIntegrityException; + +import org.apache.commons.lang3.exception.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.io.Serializable; +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; + +import static org.apache.flink.connector.kafka.util.AdminUtils.getTopicMetadata; +import static org.apache.flink.connector.kafka.util.AdminUtils.getTopicsByPattern; + +/** Provider of topic integrity related functionalities for {@link KafkaSourceEnumerator}. */ +class TopicIntegrityProvider implements Serializable { + + private static final Logger LOG = LoggerFactory.getLogger(TopicIntegrityProvider.class); + private final Map<String, String> topicIntegrityMapping = new ConcurrentHashMap<>(); + + TopicIntegrityProvider() {} + + public void open(Map<String, String> topicIntegrityMappingFromContext) { Review Comment: Maybe it was discussed offline, but could you please remind me why we can't (or don't want to) use `org.apache.kafka.common.Uuid` instead of String there (In Map<String, String> topicIntegrityMappingFromContext) ? ########## flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/subscriber/TopicIntegrityProvider.java: ########## @@ -0,0 +1,162 @@ +/* + * 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.subscriber; + +import org.apache.flink.connector.kafka.integrity.TopicIntegrityException; + +import org.apache.commons.lang3.exception.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.io.Serializable; +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; + +import static org.apache.flink.connector.kafka.util.AdminUtils.getTopicMetadata; +import static org.apache.flink.connector.kafka.util.AdminUtils.getTopicsByPattern; + +/** Provider of topic integrity related functionalities for {@link KafkaSourceEnumerator}. */ +class TopicIntegrityProvider implements Serializable { + + private static final Logger LOG = LoggerFactory.getLogger(TopicIntegrityProvider.class); + private final Map<String, String> topicIntegrityMapping = new ConcurrentHashMap<>(); + + TopicIntegrityProvider() {} + + public void open(Map<String, String> topicIntegrityMappingFromContext) { Review Comment: The method takes a raw Map(not a context), it acquires no resource, it only does putAll. That's not "opening" anything. Does not it feel weird to have such method on Provider? I understand that it is established on the Subscriber Interfaces, but not here. If I follow the idea, maybe it can be "restore/init"? ########## flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/subscriber/TopicIntegrityProvider.java: ########## @@ -0,0 +1,162 @@ +/* + * 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.subscriber; + +import org.apache.flink.connector.kafka.integrity.TopicIntegrityException; + +import org.apache.commons.lang3.exception.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.io.Serializable; +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; + +import static org.apache.flink.connector.kafka.util.AdminUtils.getTopicMetadata; +import static org.apache.flink.connector.kafka.util.AdminUtils.getTopicsByPattern; + +/** Provider of topic integrity related functionalities for {@link KafkaSourceEnumerator}. */ +class TopicIntegrityProvider implements Serializable { + + private static final Logger LOG = LoggerFactory.getLogger(TopicIntegrityProvider.class); + private final Map<String, String> topicIntegrityMapping = new ConcurrentHashMap<>(); + + TopicIntegrityProvider() {} + + public void open(Map<String, String> topicIntegrityMappingFromContext) { Review Comment: Did you consider the design where we initialize this map in TopicIntegrityProvider constructor? ########## flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/subscriber/TopicIntegrityProviderTest.java: ########## @@ -0,0 +1,229 @@ +/* + * 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.subscriber; + +import org.apache.flink.connector.kafka.integrity.TopicIntegrityException; + +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}. + * + * <p>{@link TopicIntegrityProvider#getVerifiedTopicMetadata} talks to a real {@link + * org.apache.kafka.clients.admin.AdminClient}, so these tests drive it through {@link + * TestAdminClient}, a thin wrapper around Kafka's own {@link MockAdminClient} test double. + */ +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(); + provider.open(Map.of(TOPIC1, id)); + + Map<String, TopicDescription> result = + provider.getVerifiedTopicMetadata(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(); + provider.open(new HashMap<>()); + + provider.getVerifiedTopicMetadata(mockAdmin, Collections.singletonList(TOPIC1)); + + assertThat(provider.getTopicIntegrityMapping()).containsExactly(entry(TOPIC1, id)); + } + + @Test + void testFailsIfTopicIsMissing() { + TopicIntegrityProvider provider = new TopicIntegrityProvider(); + provider.open(Map.of(TOPIC1, Uuid.randomUuid().toString())); + + assertThatThrownBy( + () -> + provider.getVerifiedTopicMetadata( + 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(); + provider.open(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.getVerifiedTopicMetadata( + mockAdmin, Collections.singletonList(TOPIC1))) + .isInstanceOf(TopicIntegrityException.class) + .hasMessageContaining("Topic " + TOPIC1 + " was recreated"); + } + + @Test + void testThrowsOriginalErrorWhenUnknownTopicExceptionIsNotDueToMissingTopic() throws Exception { + String id = addTopic(TOPIC1); + // Marked-for-deletion topics fail describeTopics with UnknownTopicOrPartitionException, + // but MockAdminClient's listTopics still reports them - a stand-in for a topic that is + // transiently unavailable rather than truly gone. + mockAdmin.markTopicForDeletion(TOPIC1); + + TopicIntegrityProvider provider = new TopicIntegrityProvider(); + provider.open(Map.of(TOPIC1, id)); + + assertThatThrownBy( + () -> + provider.getVerifiedTopicMetadata( + 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(); + provider.open(Map.of(TOPIC1, id)); + + assertThatThrownBy( + () -> + provider.getVerifiedTopicMetadata( + mockAdmin, Collections.singletonList(TOPIC1))) + .isNotInstanceOf(TopicIntegrityException.class) + .hasRootCauseInstanceOf(TimeoutException.class); + } + + @Test + void testRemovesOutdatedTopicFromMapping() throws Exception { + // TestAdminClient adminClient = new TestAdminClient(); + String id1 = addTopic(TOPIC1); + + TopicIntegrityProvider provider = new TopicIntegrityProvider(); + Map<String, String> tracked = new HashMap<>(); + tracked.put(TOPIC1, id1); + tracked.put(TOPIC2, Uuid.randomUuid().toString()); + provider.open(tracked); + + // Only TOPIC1 is subscribed to anymore; TOPIC2 must be dropped from the tracked mapping. + provider.getVerifiedTopicMetadata(mockAdmin, Collections.singletonList(TOPIC1)); + + assertThat(provider.getTopicIntegrityMapping()).containsExactly(entry(TOPIC1, id1)); + } + + @Test + void testPatternModeStillChecksTopicsThatDisappearedFromLiveMatch() { + // TOPIC1 was tracked in a previous run but no longer exists on the cluster at all, so a + // pattern-driven re-query alone would silently drop it from the set of names to verify. + // MockAdminClient adminClient = new MockAdminClient(); Review Comment: leftover? ########## flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/subscriber/TopicIntegrityProviderTest.java: ########## @@ -0,0 +1,229 @@ +/* + * 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.subscriber; + +import org.apache.flink.connector.kafka.integrity.TopicIntegrityException; + +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}. + * + * <p>{@link TopicIntegrityProvider#getVerifiedTopicMetadata} talks to a real {@link + * org.apache.kafka.clients.admin.AdminClient}, so these tests drive it through {@link + * TestAdminClient}, a thin wrapper around Kafka's own {@link MockAdminClient} test double. + */ +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(); + provider.open(Map.of(TOPIC1, id)); + + Map<String, TopicDescription> result = + provider.getVerifiedTopicMetadata(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(); + provider.open(new HashMap<>()); + + provider.getVerifiedTopicMetadata(mockAdmin, Collections.singletonList(TOPIC1)); + + assertThat(provider.getTopicIntegrityMapping()).containsExactly(entry(TOPIC1, id)); + } + + @Test + void testFailsIfTopicIsMissing() { + TopicIntegrityProvider provider = new TopicIntegrityProvider(); + provider.open(Map.of(TOPIC1, Uuid.randomUuid().toString())); + + assertThatThrownBy( + () -> + provider.getVerifiedTopicMetadata( + 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(); + provider.open(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.getVerifiedTopicMetadata( + mockAdmin, Collections.singletonList(TOPIC1))) + .isInstanceOf(TopicIntegrityException.class) + .hasMessageContaining("Topic " + TOPIC1 + " was recreated"); + } + + @Test + void testThrowsOriginalErrorWhenUnknownTopicExceptionIsNotDueToMissingTopic() throws Exception { + String id = addTopic(TOPIC1); + // Marked-for-deletion topics fail describeTopics with UnknownTopicOrPartitionException, + // but MockAdminClient's listTopics still reports them - a stand-in for a topic that is + // transiently unavailable rather than truly gone. + mockAdmin.markTopicForDeletion(TOPIC1); + + TopicIntegrityProvider provider = new TopicIntegrityProvider(); + provider.open(Map.of(TOPIC1, id)); + + assertThatThrownBy( + () -> + provider.getVerifiedTopicMetadata( + 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(); + provider.open(Map.of(TOPIC1, id)); + + assertThatThrownBy( + () -> + provider.getVerifiedTopicMetadata( + mockAdmin, Collections.singletonList(TOPIC1))) + .isNotInstanceOf(TopicIntegrityException.class) + .hasRootCauseInstanceOf(TimeoutException.class); + } + + @Test + void testRemovesOutdatedTopicFromMapping() throws Exception { + // TestAdminClient adminClient = new TestAdminClient(); + String id1 = addTopic(TOPIC1); + + TopicIntegrityProvider provider = new TopicIntegrityProvider(); + Map<String, String> tracked = new HashMap<>(); + tracked.put(TOPIC1, id1); + tracked.put(TOPIC2, Uuid.randomUuid().toString()); + provider.open(tracked); + + // Only TOPIC1 is subscribed to anymore; TOPIC2 must be dropped from the tracked mapping. + provider.getVerifiedTopicMetadata(mockAdmin, Collections.singletonList(TOPIC1)); + + assertThat(provider.getTopicIntegrityMapping()).containsExactly(entry(TOPIC1, id1)); + } + + @Test + void testPatternModeStillChecksTopicsThatDisappearedFromLiveMatch() { + // TOPIC1 was tracked in a previous run but no longer exists on the cluster at all, so a + // pattern-driven re-query alone would silently drop it from the set of names to verify. + // MockAdminClient adminClient = new MockAdminClient(); + + TopicIntegrityProvider provider = new TopicIntegrityProvider(); + provider.open(Map.of(TOPIC1, Uuid.randomUuid().toString())); + + assertThatThrownBy( + () -> provider.getVerifiedTopicMetadata(mockAdmin, Pattern.compile(".*"))) + .isInstanceOf(TopicIntegrityException.class) + .hasMessageContaining("Topic " + TOPIC1 + " is missing"); + } + + @Test + void testGetTopicIntegrityMappingReturnsDefensiveCopy() { + TopicIntegrityProvider provider = new TopicIntegrityProvider(); + provider.open(Map.of(TOPIC1, Uuid.randomUuid().toString())); + + Map<String, String> mapping = provider.getTopicIntegrityMapping(); + mapping.put(TOPIC2, Uuid.randomUuid().toString()); + + assertThat(provider.getTopicIntegrityMapping()).doesNotContainKey(TOPIC2); + } + + @Test + void testEmptySubscriptionReturnsEmptyMetadataWithoutError() { + TopicIntegrityProvider provider = new TopicIntegrityProvider(); + provider.open(new HashMap<>()); + + Map<String, TopicDescription> result = + provider.getVerifiedTopicMetadata(mockAdmin, Collections.emptyList()); + + assertThat(result).isEmpty(); + assertThat(provider.getTopicIntegrityMapping()).isEmpty(); + } + + private static final 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(); + } + + /** Adds {@code name} with a fresh topic id and returns that id. */ Review Comment: leftover? ########## flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/subscriber/TopicIntegrityProvider.java: ########## @@ -0,0 +1,162 @@ +/* + * 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.subscriber; + +import org.apache.flink.connector.kafka.integrity.TopicIntegrityException; + +import org.apache.commons.lang3.exception.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.io.Serializable; +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; + +import static org.apache.flink.connector.kafka.util.AdminUtils.getTopicMetadata; +import static org.apache.flink.connector.kafka.util.AdminUtils.getTopicsByPattern; + +/** Provider of topic integrity related functionalities for {@link KafkaSourceEnumerator}. */ +class TopicIntegrityProvider implements Serializable { + + private static final Logger LOG = LoggerFactory.getLogger(TopicIntegrityProvider.class); + private final Map<String, String> topicIntegrityMapping = new ConcurrentHashMap<>(); + + TopicIntegrityProvider() {} + + public void open(Map<String, String> topicIntegrityMappingFromContext) { + topicIntegrityMapping.putAll(topicIntegrityMappingFromContext); + } + + public Map<String, TopicDescription> getVerifiedTopicMetadata( + AdminClient adminClient, Pattern pattern) { + final Collection<String> topicsToVerifyInPatternMode = + topicIntegrityMapping.keySet().stream() + .filter(pattern.asPredicate()) + .collect(Collectors.toCollection(HashSet::new)); + topicsToVerifyInPatternMode.addAll(getTopicsByPattern(adminClient, pattern)); + return getVerifiedTopicMetadata(adminClient, topicsToVerifyInPatternMode); + } + + public Map<String, TopicDescription> getVerifiedTopicMetadata( + AdminClient adminClient, Collection<String> subscribedTopicNames) { + Map<String, TopicDescription> topicMetadata = new HashMap<>(); Review Comment: > new HashMap<>(); redundant? ########## flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/subscriber/TopicIntegrityProviderTest.java: ########## @@ -0,0 +1,229 @@ +/* + * 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.subscriber; + +import org.apache.flink.connector.kafka.integrity.TopicIntegrityException; + +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}. + * + * <p>{@link TopicIntegrityProvider#getVerifiedTopicMetadata} talks to a real {@link + * org.apache.kafka.clients.admin.AdminClient}, so these tests drive it through {@link + * TestAdminClient}, a thin wrapper around Kafka's own {@link MockAdminClient} test double. + */ +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(); + provider.open(Map.of(TOPIC1, id)); + + Map<String, TopicDescription> result = + provider.getVerifiedTopicMetadata(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(); + provider.open(new HashMap<>()); + + provider.getVerifiedTopicMetadata(mockAdmin, Collections.singletonList(TOPIC1)); + + assertThat(provider.getTopicIntegrityMapping()).containsExactly(entry(TOPIC1, id)); + } + + @Test + void testFailsIfTopicIsMissing() { + TopicIntegrityProvider provider = new TopicIntegrityProvider(); + provider.open(Map.of(TOPIC1, Uuid.randomUuid().toString())); + + assertThatThrownBy( + () -> + provider.getVerifiedTopicMetadata( + 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(); + provider.open(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.getVerifiedTopicMetadata( + mockAdmin, Collections.singletonList(TOPIC1))) + .isInstanceOf(TopicIntegrityException.class) + .hasMessageContaining("Topic " + TOPIC1 + " was recreated"); + } + + @Test + void testThrowsOriginalErrorWhenUnknownTopicExceptionIsNotDueToMissingTopic() throws Exception { + String id = addTopic(TOPIC1); + // Marked-for-deletion topics fail describeTopics with UnknownTopicOrPartitionException, + // but MockAdminClient's listTopics still reports them - a stand-in for a topic that is + // transiently unavailable rather than truly gone. + mockAdmin.markTopicForDeletion(TOPIC1); + + TopicIntegrityProvider provider = new TopicIntegrityProvider(); + provider.open(Map.of(TOPIC1, id)); + + assertThatThrownBy( + () -> + provider.getVerifiedTopicMetadata( + 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(); + provider.open(Map.of(TOPIC1, id)); + + assertThatThrownBy( + () -> + provider.getVerifiedTopicMetadata( + mockAdmin, Collections.singletonList(TOPIC1))) + .isNotInstanceOf(TopicIntegrityException.class) + .hasRootCauseInstanceOf(TimeoutException.class); + } + + @Test + void testRemovesOutdatedTopicFromMapping() throws Exception { + // TestAdminClient adminClient = new TestAdminClient(); Review Comment: leftover? ########## flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/subscriber/TopicIntegrityProvider.java: ########## @@ -0,0 +1,162 @@ +/* + * 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.subscriber; + +import org.apache.flink.connector.kafka.integrity.TopicIntegrityException; + +import org.apache.commons.lang3.exception.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.io.Serializable; +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; + +import static org.apache.flink.connector.kafka.util.AdminUtils.getTopicMetadata; +import static org.apache.flink.connector.kafka.util.AdminUtils.getTopicsByPattern; + +/** Provider of topic integrity related functionalities for {@link KafkaSourceEnumerator}. */ +class TopicIntegrityProvider implements Serializable { + + private static final Logger LOG = LoggerFactory.getLogger(TopicIntegrityProvider.class); + private final Map<String, String> topicIntegrityMapping = new ConcurrentHashMap<>(); + + TopicIntegrityProvider() {} + + public void open(Map<String, String> topicIntegrityMappingFromContext) { + topicIntegrityMapping.putAll(topicIntegrityMappingFromContext); + } + + public Map<String, TopicDescription> getVerifiedTopicMetadata( + AdminClient adminClient, Pattern pattern) { + final Collection<String> topicsToVerifyInPatternMode = + topicIntegrityMapping.keySet().stream() + .filter(pattern.asPredicate()) + .collect(Collectors.toCollection(HashSet::new)); + topicsToVerifyInPatternMode.addAll(getTopicsByPattern(adminClient, pattern)); + return getVerifiedTopicMetadata(adminClient, topicsToVerifyInPatternMode); + } + + public Map<String, TopicDescription> getVerifiedTopicMetadata( + AdminClient adminClient, Collection<String> subscribedTopicNames) { + Map<String, TopicDescription> topicMetadata = new HashMap<>(); + try { + topicMetadata = getTopicMetadata(adminClient, subscribedTopicNames); + failIfRecreated(subscribedTopicNames, topicMetadata); + } catch (RuntimeException original) { + if (ExceptionUtils.getRootCause(original) instanceof UnknownTopicOrPartitionException) { + // 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 (Exception ignored) { + // ignored so we fallback to the original error + } + } + throw original; + } + trackNewIdsInMapping(subscribedTopicNames, topicMetadata); + return topicMetadata; + } + + private void trackNewIdsInMapping( + Collection<String> subscribedTopicNames, Map<String, TopicDescription> topicMetadata) { + + // Add new subscribed topic to mapping + for (String subscribedTopicName : subscribedTopicNames) { + if (!topicIntegrityMapping.keySet().contains(subscribedTopicName)) { + topicIntegrityMapping.put( + subscribedTopicName, + topicMetadata.get(subscribedTopicName).topicId().toString()); + } + } + // Remove outdated topics from mapping + for (String topicNameFromMapping : topicIntegrityMapping.keySet()) { + if (!subscribedTopicNames.contains(topicNameFromMapping)) { + topicIntegrityMapping.remove(topicNameFromMapping); + } + } + } + + public Map<String, String> getTopicIntegrityMapping() { + return new HashMap<>(topicIntegrityMapping); + } + + 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 = topicIntegrityMapping.get(subscribedTopicName); + final Uuid topicIdFromMetadata = topicDescription.topicId(); + if (topicIdFromState == null || topicIdFromMetadata == null) { + // we skip topic integrity check for null topicId + // due to broker configuration, or topic not yet stored on topicIntegrityMapping + 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); + return; Review Comment: this return ends the loop. is it by design? If one topic has a null id, the return skips the integrity check for every remaining subscribed topic, not just that one. ########## flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/subscriber/TopicIntegrityProvider.java: ########## @@ -0,0 +1,162 @@ +/* + * 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.subscriber; + +import org.apache.flink.connector.kafka.integrity.TopicIntegrityException; + +import org.apache.commons.lang3.exception.ExceptionUtils; Review Comment: why not org.apache.flink.util.ExceptionUtils ? ########## flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/subscriber/TopicIntegrityProviderTest.java: ########## @@ -0,0 +1,229 @@ +/* + * 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.subscriber; + +import org.apache.flink.connector.kafka.integrity.TopicIntegrityException; + +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}. + * + * <p>{@link TopicIntegrityProvider#getVerifiedTopicMetadata} talks to a real {@link + * org.apache.kafka.clients.admin.AdminClient}, so these tests drive it through {@link + * TestAdminClient}, a thin wrapper around Kafka's own {@link MockAdminClient} test double. Review Comment: The link TestAdminClient is broken in my IDE when I review it. probably this class was removed during refactorings? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
