litiliu commented on code in PR #4608:
URL: https://github.com/apache/flink-cdc/pull/4608#discussion_r4535026551
##########
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()
+ .filter(info ->
info.physicalTablePath.equals(partitionPath))
+ .map(info -> info.tableBucket)
+ .forEach(bucket ->
pendingPartitionRemovalBuckets.put(bucket, partitionPath));
+ LOG.warn(
+ "Removing partition {} from the source. KV snapshot leases
are retained until their existing expiry or close handling.",
+ partitionPath);
+ }
+ assignedPhysicalTablePaths.removeIf(
+ path ->
+ path.getPartitionName() != null
+ &&
pendingPartitionRemovalRequests.containsKey(path));
+ pendingPartitionSplitAssignment
+ .values()
+ .forEach(
+ splits ->
+ splits.removeIf(
+ split ->
+ !isDiscoveredPartitionBucket(
Review Comment:
Changed the inventory to use `partitionInfo.getBucketCount()`. Added tests
to cover both changing the table default from 4 to 2 before the first discovery
and changing it after the four splits are pending but before readers register.
In both cases, the existing partition keeps buckets 0–3 and all four are
assigned.
--
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]
