Savonitar commented on code in PR #287:
URL:
https://github.com/apache/flink-connector-kafka/pull/287#discussion_r3871688251
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilderTest.java:
##########
@@ -273,6 +273,58 @@ public void testPeriodPartitionDiscovery() {
.isEqualTo(-1L);
}
+ @Test
+ public void testDefaultCheckSourceIntegrity() {
+ final KafkaSource<String> kafkaSource = getBasicBuilder().build();
+ // Commit on checkpoint and auto commit should be disabled because
group.id is not specified
Review Comment:
Looks like a leftover? Or is it an intentional place for this comment?
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java:
##########
@@ -213,6 +219,25 @@ public KafkaSourceEnumerator(
this.initialDiscoveryFinished =
kafkaSourceEnumState.initialDiscoveryFinished();
this.assignedSplits =
indexByPartition(kafkaSourceEnumState.assignedSplits());
this.unassignedSplits =
indexByPartition(kafkaSourceEnumState.unassignedSplits());
+ this.topicIntegrityCheckEnabled =
+ KafkaSourceOptions.getOption(
+ properties,
+ KafkaSourceOptions.TOPIC_INTEGRITY_CHECK_ENABLED,
+ Boolean::parseBoolean);
+ this.topicIntegrityProvider =
+ new TopicIntegrityProvider(
+ topicIntegrityCheckEnabled
+ ? kafkaSourceEnumState.trackedTopicIdsByName()
+ : Collections.emptyMap());
+ LOG.debug(
+ "KafkaSourceEnumerator initialized with assignedSplits: {},
unassignedSplits: {}, "
+ + "initialDiscoveryFinished: {},
topicIntegrityCheckEnabled: {}, trackedTopicIdsByName: {}, properties: {}",
+ assignedSplits.keySet(),
+ unassignedSplits.keySet(),
+ initialDiscoveryFinished,
+ topicIntegrityCheckEnabled,
+ topicIntegrityProvider.getTrackedTopicIdsByName(),
+ properties);
Review Comment:
Here we log properties. They can contain password which will be the security
issue.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceOptions.java:
##########
@@ -70,6 +70,13 @@ public class KafkaSourceOptions {
+ "of polling more often. 0 polls without
blocking. Must not be "
+ "negative.");
+ public static final ConfigOption<Boolean> TOPIC_INTEGRITY_CHECK_ENABLED =
+ ConfigOptions.key("scan.topic-integrity-check.enabled")
Review Comment:
do we need to add this option to
[KafkaConnectorOptions](https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaConnectorOptions.java)?
If no:
1. FLIP
https://cwiki.apache.org/confluence/spaces/FLINK/pages/406619238/FLIP-562+Topic+integrity+checks+in+Kafka+Connector#FLIP562%3ATopicintegritychecksinKafkaConnector-KafkaConnectorOptions.java%3A
mentions that
2. Will users be able to specify this option in SQL query?
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumeratorTest.java:
##########
@@ -947,4 +947,46 @@ private void runPeriodicPartitionDiscovery(
context.runNextOneTimeCallable();
}
}
+
+ @Test
+ public void testCheckSourceIntegrityFromProperties() throws Exception {
+ // Test that properties are used when job configuration doesn't have
the setting
+ final boolean propertiesCheckSourceIntegrity = true;
+
+ Properties properties = new Properties();
+ properties.setProperty(
+ KafkaSourceOptions.TOPIC_INTEGRITY_CHECK_ENABLED.key(),
+ String.valueOf(propertiesCheckSourceIntegrity));
+ try (MockSplitEnumeratorContext<KafkaPartitionSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS);
+ KafkaSourceEnumerator enumerator =
+ createEnumerator(
+ context,
+ ENABLE_PERIODIC_PARTITION_DISCOVERY ? 1 : -1,
+ OffsetsInitializer.earliest(),
+ Collections.emptySet(),
+ Collections.emptySet(),
+ Collections.emptySet(),
+ true,
+ properties)) {
+
+ // Verify that the properties value is used
+ assertThat(propertiesCheckSourceIntegrity)
+ .isEqualTo(enumerator.topicIntegrityCheckEnabled());
+ }
+ }
+
+ @Test
+ public void testCheckSourceIntegrityDefaultValue() throws Exception {
+ final boolean defaultCheckSourceIntegrity = false;
+ try (MockSplitEnumeratorContext<KafkaPartitionSplit> context =
+ new MockSplitEnumeratorContext<>(NUM_SUBTASKS);
+ KafkaSourceEnumerator enumerator =
+ createEnumerator(context,
DISABLE_PERIODIC_PARTITION_DISCOVERY)) {
+
+ // Verify partition discovery is disabled
Review Comment:
typo? seems like it should mention topicIntegrity, not partitionDiscovery?
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicMetadataSettable.java:
##########
@@ -0,0 +1,25 @@
+/*
+ * 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;
+
+/** Interface for setting a custom {@link TopicMetadataProvider} other than
the default. */
+public interface TopicMetadataSettable {
Review Comment:
Thanks for adding `@Internal ` to TopicIntegrityProvider and
TopicMetadataProvider. Should we also add this annotation here?
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceOptions.java:
##########
@@ -70,6 +70,13 @@ public class KafkaSourceOptions {
+ "of polling more often. 0 polls without
blocking. Must not be "
+ "negative.");
+ public static final ConfigOption<Boolean> TOPIC_INTEGRITY_CHECK_ENABLED =
Review Comment:
Should we add this option to docs?
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicIntegrityProvider.java:
##########
@@ -0,0 +1,175 @@
+/*
+ * 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.keySet().contains(subscribedTopicName))
{
+ if (topicMetadata.get(subscribedTopicName).topicId() == null
Review Comment:
Nit: I see this get() method
```
topicMetadata.get(subscribedTopicName)
```
is executed 3 times in same if-branch. does it make sense to extract it?
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicIntegrityProvider.java:
##########
@@ -0,0 +1,175 @@
+/*
+ * 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.keySet().contains(subscribedTopicName))
{
Review Comment:
nit: containsKey
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicIntegrityException.java:
##########
@@ -0,0 +1,27 @@
+/*
+ * 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.util.FlinkRuntimeException;
+
+/** Exception thrown when topic integrity check fails. */
+public class TopicIntegrityException extends FlinkRuntimeException {
Review Comment:
Is it unrecoverable exception?
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/metadata/TopicIntegrityException.java:
##########
@@ -0,0 +1,27 @@
+/*
+ * 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.util.FlinkRuntimeException;
+
+/** Exception thrown when topic integrity check fails. */
+public class TopicIntegrityException extends FlinkRuntimeException {
Review Comment:
Thanks for adding `@Internal ` to TopicIntegrityProvider and
TopicMetadataProvider. Should we also add this annotation here?
--
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]
