tzulitai commented on a change in pull request #7726: [FLINK-10342] Filter
restored partitions with discovered partitions b…
URL: https://github.com/apache/flink/pull/7726#discussion_r259889775
##########
File path:
flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
##########
@@ -494,6 +514,12 @@ public void open(Configuration configuration) throws
Exception {
}
}
+ if (filterRestoredPartitionsWithDiscovered) {
+
subscribedPartitionsToStartOffsets.entrySet().removeIf(
+ entry ->
(!allPartitions.contains(entry.getKey()))
Review comment:
I'm not entirely sure this is safe.
This is relying on the fact that Kafka indeed returns the correct set of
partitions.
While in normal conditions, the end results should be the same, but I feel
like it is more safer to filter based on just the `topicsDescriptor` so we're
not relying on external system behaviours and subject to any system hiccups.
@fengli What do you think?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
