litiliu commented on code in PR #4608:
URL: https://github.com/apache/flink-cdc/pull/4608#discussion_r4535027826
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/enumerator/FlussSourceEnumerator.java:
##########
@@ -298,15 +316,96 @@ private void checkTableBucketChanges(DiscoveryResult
discoveryResult, Throwable
// Removal cleanup needs the previous discovery's buckets, absent from
the new result.
boolean subscriptionChanged =
updateSubscriptions(discoveryResult.subscribedTablePaths);
+ Set<TableBucket> currentBuckets =
+ discoveryResult.tableBuckets.stream()
+ .map(info -> info.tableBucket)
+ .collect(Collectors.toSet());
+ boolean bucketsChanged =
!discoveredTableBuckets.equals(currentBuckets);
+ discoveredTableBuckets.clear();
+ discoveredTableBuckets.addAll(currentBuckets);
+ discoveredPhysicalTablePaths.clear();
+ discoveryResult.tableBuckets.stream()
+ .map(info -> info.physicalTablePath)
+ .forEach(discoveredPhysicalTablePaths::add);
+ boolean partitionsRemoved = removeExpiredPartitions();
lastDiscoveredTableBuckets = discoveryResult.tableBuckets;
- if (!sentSubscriptionSnapshot || subscriptionChanged) {
+ if (!sentSubscriptionSnapshot
+ || subscriptionChanged
+ || bucketsChanged
+ || partitionsRemoved) {
sendSubscriptionSnapshot();
sentSubscriptionSnapshot = true;
}
initializeNewTableBuckets(discoveryResult.tableBuckets);
}
+ private boolean removeExpiredPartitions() {
+ Set<PhysicalTablePath> removedPartitions = new
HashSet<>(assignedPhysicalTablePaths);
+ removedPartitions.addAll(initializingPhysicalTablePaths);
+ pendingPartitionSplitAssignment.values().stream()
+ .flatMap(Set::stream)
+ .map(FlussSplitBase::getPhysicalTablePath)
+ .forEach(removedPartitions::add);
+ lastDiscoveredTableBuckets.stream()
+ .map(info -> info.physicalTablePath)
+ .forEach(removedPartitions::add);
+ removedPartitions.removeIf(
+ path ->
+ path.getPartitionName() == null
+ ||
!subscribedTablePaths.contains(path.getTablePath())
+ ||
pendingRemovalTablePaths.contains(path.getTablePath())
+ || discoveredPhysicalTablePaths.contains(path)
+ ||
pendingPartitionRemovalRequests.containsKey(path));
+ for (PhysicalTablePath partitionPath : removedPartitions) {
+ pendingPartitionRemovalRequests.put(
+ partitionPath, NEXT_REMOVAL_REQUEST_ID.incrementAndGet());
+ partitionRemovalAcknowledgements.remove(partitionPath);
+ lastDiscoveredTableBuckets.stream()
Review Comment:
Applied the single-pass approach. After collecting `removedPartitions`, the
bucket-mapping step scans `lastDiscoveredTableBuckets` once and checks
`removedPartitions.contains(info.physicalTablePath)`, instead of rescanning the
full list for each expired partition.
--
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]
