Savonitar commented on code in PR #334: URL: https://github.com/apache/flink-connector-kafka/pull/334#discussion_r4474205402
########## flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorTopicIntegrityTest.java: ########## @@ -0,0 +1,320 @@ +/* + * 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.dynamic.source.enumerator; + +import org.apache.flink.api.connector.source.Boundedness; +import org.apache.flink.api.connector.source.ReaderInfo; +import org.apache.flink.api.connector.source.SplitEnumeratorContext; +import org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext; +import org.apache.flink.connector.kafka.dynamic.metadata.KafkaMetadataService; +import org.apache.flink.connector.kafka.dynamic.metadata.KafkaStream; +import org.apache.flink.connector.kafka.dynamic.source.DynamicKafkaSourceOptions; +import org.apache.flink.connector.kafka.dynamic.source.GetMetadataUpdateEvent; +import org.apache.flink.connector.kafka.dynamic.source.enumerator.subscriber.KafkaStreamSetSubscriber; +import org.apache.flink.connector.kafka.dynamic.source.split.DynamicKafkaSourceSplit; +import org.apache.flink.connector.kafka.source.KafkaSourceOptions; +import org.apache.flink.connector.kafka.source.enumerator.AssignmentStatus; +import org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumState; +import org.apache.flink.connector.kafka.source.enumerator.SplitAndAssignmentStatus; +import org.apache.flink.connector.kafka.source.enumerator.initializer.NoStoppingOffsetsInitializer; +import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer; +import org.apache.flink.connector.kafka.source.enumerator.metadata.TopicIntegrityException; +import org.apache.flink.connector.kafka.testutils.MockKafkaMetadataService; +import org.apache.flink.streaming.connectors.kafka.DynamicKafkaSourceTestHelper; + +import org.apache.kafka.common.Uuid; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.Set; +import java.util.function.Consumer; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for the topic integrity check of the {@link DynamicKafkaSourceEnumerator}. */ +public class DynamicKafkaSourceEnumeratorTopicIntegrityTest { Review Comment: Thanks for the split 👍 ########## flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/StoppableKafkaEnumContextProxyTest.java: ########## @@ -131,6 +133,61 @@ public void testPeriodicCallableThrowsExceptionOnActiveCluster() throws Throwabl } } + @Test + public void testTopicIntegrityErrorSwallowedOnInactiveCluster() throws Throwable { + try (MockSplitEnumeratorContext<DynamicKafkaSourceSplit> enumContext = + new MockSplitEnumeratorContext<>(2); + StoppableKafkaEnumContextProxy enumContextProxy = + createStoppableKafkaEnumContextProxy(enumContext)) { + + AtomicBoolean isCallbackInvoked = new AtomicBoolean(); + setupKafkaTopicPartitionDiscoveryMockCallable( + enumContextProxy, + isCallbackInvoked, + () -> new TopicIntegrityException("Topic mock-topic was recreated")); + + CommonTestUtils.waitUtil( + () -> enumContext.getOneTimeCallables().size() == 1, Review Comment: Could we wait for `!enumContext.getOneTimeCallables().isEmpty()` here and in the active-cluster test? The helper enqueues every second. If two calls accumulate before this wait, size() == 1 cannot become true without consuming a call, so the test times out If I'm not mistaken. ########## docs/content/docs/connectors/table/dynamic-kafka.md: ########## @@ -182,6 +182,14 @@ Connector Options <td>Integer</td> <td>Number of consecutive discovery failures before failing the job.</td> </tr> + <tr> + <td><h5>scan.topic-integrity-check.enabled</h5></td> + <td>optional</td> + <td>no</td> + <td style="word-wrap: break-word;">false</td> + <td>Boolean</td> + <td>Whether to verify the id of the topics subscribed on each Kafka cluster during runtime and fail the job if a topic is missing or was recreated. The check is performed on startup and, if <code>scan.topic-partition-discovery.interval</code> is enabled, on every discovery interval.</td> Review Comment: > scan.topic-partition-discovery.interval not blocking anything, but should this option also be added to this page? ########## flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorTopicIntegrityTest.java: ########## @@ -0,0 +1,320 @@ +/* + * 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.dynamic.source.enumerator; + +import org.apache.flink.api.connector.source.Boundedness; +import org.apache.flink.api.connector.source.ReaderInfo; +import org.apache.flink.api.connector.source.SplitEnumeratorContext; +import org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext; +import org.apache.flink.connector.kafka.dynamic.metadata.KafkaMetadataService; +import org.apache.flink.connector.kafka.dynamic.metadata.KafkaStream; +import org.apache.flink.connector.kafka.dynamic.source.DynamicKafkaSourceOptions; +import org.apache.flink.connector.kafka.dynamic.source.GetMetadataUpdateEvent; +import org.apache.flink.connector.kafka.dynamic.source.enumerator.subscriber.KafkaStreamSetSubscriber; +import org.apache.flink.connector.kafka.dynamic.source.split.DynamicKafkaSourceSplit; +import org.apache.flink.connector.kafka.source.KafkaSourceOptions; +import org.apache.flink.connector.kafka.source.enumerator.AssignmentStatus; +import org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumState; +import org.apache.flink.connector.kafka.source.enumerator.SplitAndAssignmentStatus; +import org.apache.flink.connector.kafka.source.enumerator.initializer.NoStoppingOffsetsInitializer; +import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer; +import org.apache.flink.connector.kafka.source.enumerator.metadata.TopicIntegrityException; +import org.apache.flink.connector.kafka.testutils.MockKafkaMetadataService; +import org.apache.flink.streaming.connectors.kafka.DynamicKafkaSourceTestHelper; + +import org.apache.kafka.common.Uuid; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.Set; +import java.util.function.Consumer; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for the topic integrity check of the {@link DynamicKafkaSourceEnumerator}. */ +public class DynamicKafkaSourceEnumeratorTopicIntegrityTest { + private static final int NUM_SUBTASKS = 3; + private static final String TOPIC = "DynamicKafkaSourceEnumeratorTopicIntegrityTest"; + private static final int NUM_SPLITS_PER_CLUSTER = 3; + private static final int NUM_RECORDS_PER_SPLIT = 5; + + @BeforeAll + public static void beforeAll() throws Throwable { + DynamicKafkaSourceTestHelper.setup(); + DynamicKafkaSourceTestHelper.createTopic(TOPIC, NUM_SPLITS_PER_CLUSTER, 1); + DynamicKafkaSourceTestHelper.produceToKafka( + TOPIC, NUM_SPLITS_PER_CLUSTER, NUM_RECORDS_PER_SPLIT); + } + + @AfterAll + public static void afterAll() throws Exception { + DynamicKafkaSourceTestHelper.tearDown(); + } + + @Test + public void testTrackedTopicIdsSurviveMetadataChangeAndDetectRecreation() throws Throwable { + final String topic2 = TOPIC + "_integrity"; + + DynamicKafkaSourceTestHelper.createTopic(topic2, NUM_SPLITS_PER_CLUSTER, 1); + DynamicKafkaSourceTestHelper.produceToKafka( + topic2, NUM_SPLITS_PER_CLUSTER, NUM_RECORDS_PER_SPLIT); + final String clusterId = DynamicKafkaSourceTestHelper.getKafkaClusterId(0); + + final Set<KafkaStream> initialStreams = + Collections.singleton( + new KafkaStream( + TOPIC, + DynamicKafkaSourceTestHelper.getClusterMetadataMap( + 0, TOPIC, topic2))); + final Set<KafkaStream> updatedStreams = + Collections.singleton( + new KafkaStream( + TOPIC, + DynamicKafkaSourceTestHelper.getClusterMetadataMap(0, TOPIC))); + + final DynamicKafkaSourceEnumState stateAfterMetadataChange; + final Map<String, String> initialTrackedIds; + try (MockKafkaMetadataService metadataService = + new MockKafkaMetadataService(initialStreams); + MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context = + new MockSplitEnumeratorContext<>(NUM_SUBTASKS); + DynamicKafkaSourceEnumerator enumerator = + createEnumerator( + context, + metadataService, + DynamicKafkaSourceEnumeratorTopicIntegrityTest + ::enableTopicIntegrityCheck)) { + enumerator.start(); + context.runPeriodicCallable(0); + runAllOneTimeCallables(context); + mockRegisterReaderAndSendReaderStartupEvent(context, enumerator, 0); + mockRegisterReaderAndSendReaderStartupEvent(context, enumerator, 1); + + initialTrackedIds = getTrackedTopicIds(enumerator.snapshotState(-1), clusterId); + assertThat(initialTrackedIds).containsOnlyKeys(TOPIC, topic2); + + metadataService.setKafkaStreams(updatedStreams); + context.runPeriodicCallable(0); + + // the restarted enumerators have not rediscovered the topics yet, so the ids can only + // come from the state carried over the metadata change + assertThat(getTrackedTopicIds(enumerator.snapshotState(-1), clusterId)) + .containsOnly(Map.entry(TOPIC, initialTrackedIds.get(TOPIC))); + + // the rediscovery agrees with the carried over id, so the check passes + runAllOneTimeCallables(context); + mockRegisterReaderAndSendReaderStartupEvent(context, enumerator, 2); + + stateAfterMetadataChange = enumerator.snapshotState(-1); + assertThat(getTrackedTopicIds(stateAfterMetadataChange, clusterId)) + .containsOnly(Map.entry(TOPIC, initialTrackedIds.get(TOPIC))); + } + + // a stale id in the checkpoint is what a topic recreated after the checkpoint looks like + final KafkaSourceEnumState clusterState = + stateAfterMetadataChange.getClusterEnumeratorStates().get(clusterId); + final String staleTopicId = Uuid.randomUuid().toString(); + final DynamicKafkaSourceEnumState staleState = + withClusterState( + stateAfterMetadataChange, + clusterId, + new KafkaSourceEnumState( + clusterState.splits(), + clusterState.initialDiscoveryFinished(), + Collections.singletonMap(TOPIC, staleTopicId))); + + // topic2 is back in the stream, so the restored enumerators are restarted on the first + // metadata refresh before they run their first discovery + try (MockKafkaMetadataService metadataService = + new MockKafkaMetadataService(initialStreams); + MockSplitEnumeratorContext<DynamicKafkaSourceSplit> context = + new MockSplitEnumeratorContext<>(NUM_SUBTASKS); + DynamicKafkaSourceEnumerator enumerator = + createEnumerator( + context, + metadataService, + DynamicKafkaSourceEnumeratorTopicIntegrityTest + ::enableTopicIntegrityCheck, + staleState)) { + enumerator.start(); + context.runPeriodicCallable(0); + + assertThat(getTrackedTopicIds(enumerator.snapshotState(-1), clusterId)) + .containsOnly(Map.entry(TOPIC, staleTopicId)); + assertThatThrownBy(() -> runAllOneTimeCallables(context)) + .hasRootCauseInstanceOf(TopicIntegrityException.class) + .hasRootCauseMessage("Topic " + TOPIC + " was recreated"); + } + } + + @Test + public void testRestoreKeepsTrackedTopicIdsOfTopicsWithUnassignedSplits() throws Throwable { + final String topic2 = TOPIC + "_unassigned"; + DynamicKafkaSourceTestHelper.createTopic(topic2, NUM_SPLITS_PER_CLUSTER, 1); + DynamicKafkaSourceTestHelper.produceToKafka( + topic2, NUM_SPLITS_PER_CLUSTER, NUM_RECORDS_PER_SPLIT); + final String clusterId = DynamicKafkaSourceTestHelper.getKafkaClusterId(0); + final KafkaStream stream = + new KafkaStream( + TOPIC, + DynamicKafkaSourceTestHelper.getClusterMetadataMap(0, TOPIC, topic2)); + + final DynamicKafkaSourceEnumState checkpointState = + getCheckpointStateWithTopicIntegrityCheck(stream); + final KafkaSourceEnumState clusterState = + checkpointState.getClusterEnumeratorStates().get(clusterId); + final Map<String, String> trackedIds = clusterState.trackedTopicIdsByName(); + assertThat(trackedIds).containsOnlyKeys(TOPIC, topic2); + + // a checkpoint taken before the splits of topic2 were handed to a reader + final Set<SplitAndAssignmentStatus> splits = Review Comment: Should we also cover a restored topic with a tracked ID but without splits yet? I think a checkpoint can capture that state between metadata discovery and split initialization (correct me if im wrong). Here every tracked topic still has splits, so the test would pass even if `restoredTopics.addAll(state.trackedTopicIdsByName().keySet())` was removed. -- 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]
