MartijnVisser commented on code in PR #334:
URL:
https://github.com/apache/flink-connector-kafka/pull/334#discussion_r4371148353
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java:
##########
@@ -545,7 +545,10 @@ private void
onHandleSubscribedStreamsFetch(Set<KafkaStream> fetchedKafkaStreams
newKafkaSourceEnumState =
new KafkaSourceEnumState(
- partitions,
kafkaSourceEnumState.initialDiscoveryFinished());
+ partitions,
+
kafkaSourceEnumState.initialDiscoveryFinished(),
+ filterTrackedTopicIds(
Review Comment:
Yes, reproduced. With the check on, a deleted topic on a cluster that has
left the metadata fails the job instead of being swallowed. Also matching
`TopicIntegrityException` in `wrapCallAsyncCallable` fixes it, with a test for
that case.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java:
##########
@@ -1076,6 +1079,18 @@ private Set<SplitAndAssignmentStatus>
filterStateByTopics(
.collect(Collectors.toSet());
}
+ /**
+ * Keeps the tracked topic ids of the still active topics, so a topic
recreated while the
+ * enumerators are being restarted on a metadata change is still detected
by the topic integrity
+ * check.
+ */
+ private Map<String, String> filterTrackedTopicIds(
+ KafkaSourceEnumState kafkaSourceEnumState, Set<String>
activeTopics) {
+ return kafkaSourceEnumState.trackedTopicIdsByName().entrySet().stream()
Review Comment:
Agreed, reproduced on restore: the sub-enumerators run their discovery
before the first metadata refresh and drop that id. Taking the topics from
`state.splits()` fixes it, with the dynamic source tests still green.
##########
flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactory.java:
##########
@@ -157,6 +158,7 @@ public Set<ConfigOption<?>> optionalOptions() {
options.add(SCAN_BOUNDED_TIMESTAMP_MILLIS);
options.add(SCAN_PARALLELISM);
options.add(TRANSACTION_NAMING_STRATEGY);
+ options.add(SCAN_TOPIC_INTEGRITY_CHECK_ENABLED);
Review Comment:
Agreed, it doesn't change the topology. The Forwarded column in the docs
then becomes yes.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorTest.java:
##########
@@ -841,6 +841,76 @@ public void
testEnumeratorStateDoesNotContainStaleTopicPartitions() throws Throw
}
}
+ @Test
+ public void testTrackedTopicIdsSurviveMetadataChange() throws Throwable {
Review Comment:
I think so. Recreating the kept topic after the metadata change and
expecting "was recreated" would also cover the proxy path.
--
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]
