Savonitar commented on code in PR #334:
URL:
https://github.com/apache/flink-connector-kafka/pull/334#discussion_r4367962997
##########
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:
The test asserts that the IDs survive the round trip. Do we also need to
assert that a recreated topic is detected?
##########
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:
Does not `StoppableKafkaEnumContextProxy.wrapCallAsyncCallable` swallow
discovery-errors for a cluster that the metadata service no longer reports as
active but only if there is a `KafkaException` in the cause chain?
`TopicIntegrityException` is created with a message only, so it never
matches?
##########
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:
Should this also go into `forwardOptions()`? same as
`SCAN_TOPIC_PARTITION_DISCOVERY`
##########
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:
During restore we take "assignedSplits" only
https://github.com/Efrat19/flink-connector-kafka/blob/613caff6bc98e671b0f2b11f850bf87a83f234b8/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java#L276
shouldn't it be all `splits()` (plus the tracked ids) instead?
Because the topic whose splits were still unassigned in the checkpoint is
then not subscribed, and `TopicIntegrityProvider.refreshTrackedTopicIds` drops
its id on the first discovery.
The next metadata refresh re-adds it as a "new" topic with no tracked id, so
a recreation in between is not detected.
WDYT?
--
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]
