Savonitar commented on code in PR #287: URL: https://github.com/apache/flink-connector-kafka/pull/287#discussion_r3646864843
########## flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicIntegrityProvider.java: ########## @@ -0,0 +1,164 @@ +/* + * 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.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}. + */ +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(pattern.asPredicate()) + .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 (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.keySet().contains(subscribedTopicName)) { + trackedTopicIdsByName.put( + subscribedTopicName, + topicMetadata.get(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); + break; Review Comment: should it be continue? e.g. what if we restore from a checkpoint after a topic was added? ########## flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicIntegrityProvider.java: ########## @@ -0,0 +1,164 @@ +/* + * 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.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}. + */ +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(pattern.asPredicate()) + .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 (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.keySet().contains(subscribedTopicName)) { + trackedTopicIdsByName.put( Review Comment: should we filter out `Uuid.ZERO_UUID` before putting it in the map? otherwise, if 1.we use broker without topic id support (== returns Uuid..ZERO_UUID) 2.we store `Uuid.ZERO_UUID` in checkpoint 3.we update kafka broker to version with topic id 4.check will fail with "topic recreated", because `Uuid.ZERO_UUID` will not match real topicId. Long story short: broker update will cause failure on flink side. I didnt find this scenario in the FLIP. Is this failure by design? ########## 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: sounds good to me ########## flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java: ########## @@ -292,7 +318,10 @@ public void addReader(int subtaskId) { @Override public KafkaSourceEnumState snapshotState(long checkpointId) throws Exception { return new KafkaSourceEnumState( - assignedSplits.values(), unassignedSplits.values(), initialDiscoveryFinished); + assignedSplits.values(), + unassignedSplits.values(), + initialDiscoveryFinished, + topicIntegrityProvider.getTrackedTopicIdsByName()); Review Comment: Do we always snapshot the restored map? So ids from an earlier enabled execution are carried over even if the check is off, if a topic is recreated during that time and the user re-enables the check, the job fails on the stale id. is it intentional? ########## flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicMetadataProvider.java: ########## @@ -0,0 +1,57 @@ +/* + * 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.connector.kafka.util.AdminUtils; + +import org.apache.kafka.clients.admin.AdminClient; +import org.apache.kafka.clients.admin.TopicDescription; + +import java.io.Serializable; +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. + */ +public interface TopicMetadataProvider extends Serializable { + + Map<String, TopicDescription> getTopicMetadata( + AdminClient adminClient, Collection<String> subscribedTopicNames); + + Map<String, TopicDescription> getTopicMetadata(AdminClient adminClient, Pattern pattern); + + static TopicMetadataProvider createDefault() { + return new TopicMetadataProvider() { Review Comment: This returns anonymous class. If in the future somebody adds a new anonymous class to that file, deserialization of existed persisted job graph will fail because we do not declare `serialVersionUID` explicitly. Potential options to think about: 1. add `serialVersionUID` + make it static nested class? OR 2. make TopicMetadataProvider transient? ########## flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicIntegrityProvider.java: ########## @@ -0,0 +1,164 @@ +/* + * 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.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}. + */ +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(pattern.asPredicate()) + .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 (Exception ignored) { + // ignored so we fallback to the original error Review Comment: if adminClient.listTopics().names().get() throws InterrupedException, we will swallow it here and lose the thread's interrupt flag. Could we restore it same way as in org.apache.flink.connector.kafka.util.AdminUtils#checkIfInterrupted ? ########## 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: IMO yes -- 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]
