fxbing commented on code in PR #4608:
URL: https://github.com/apache/flink-cdc/pull/4608#discussion_r4510581468
##########
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:
Lowering the table default from 4 buckets to 2 leaves an existing 4-bucket
partition unchanged. When the reader registered after the next discovery, this
branch assigned only buckets 0 and 1. `master` assigned all four. Could we
build the inventory using `partitionInfo.getBucketCount()` and add a regression
test for this case?
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSourceReader.java:
##########
@@ -217,11 +225,23 @@ public void handleSourceEvents(SourceEvent sourceEvent) {
fetcherManager.removeTables(activeRemovalTablePaths);
}
+ Set<PhysicalTablePath> activeRemovalPartitionPaths = new HashSet<>();
+ for (FlussSplitBase split : activeSplits.values()) {
+ if (isActive(split.getTablePath())
+ &&
pendingPartitionRemovalRequests.containsKey(split.getPhysicalTablePath())) {
+ activeRemovalPartitionPaths.add(split.getPhysicalTablePath());
+ reportedFinishedSnapshotBuckets.remove(split.getTableBucket());
+ }
+ }
+ fetcherManager.removePartitions(activeRemovalPartitionPaths);
Review Comment:
After the last local split for table A finishes, a later table-removal
request finds no active splits for A and skips `removeTables()`. If another
table keeps the fetcher alive, recreating A then fails with `Table ID mismatch`
because the old `Table` is still cached. Could we clear A’s cached resources on
table removal even when it has no active splits?
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSourceReader.java:
##########
@@ -243,7 +263,7 @@ protected void onSplitFinished(Map<String, FlussSplitState>
finishedSplitIds) {
private void activateSplits(List<FlussSplitBase> splits) {
List<FlussSplitBase> activeSplits = new ArrayList<>();
for (FlussSplitBase split : splits) {
- if (isActive(split.getTablePath())) {
+ if (isSplitActive(split)) {
Review Comment:
A partition can pass this check and be dropped before the fetcher calls
`subscribe()`. With `[expired, live]` in one assignment, the SDK throws
`PartitionNotExistException` on the first split and never subscribes the
second. Could we keep the live split and trigger a fresh discovery to confirm
whether the failed split should be retired?
##########
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:
One optional performance improvement here would be to scan
`lastDiscoveredTableBuckets` once and check
`removedPartitions.contains(info.physicalTablePath)`. The current loop scans
the full bucket list once per expired partition on the coordinator thread.
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/source/reader/FlussSourceFetcherManager.java:
##########
@@ -44,25 +46,37 @@ void removeTables(Set<TablePath> tablePaths) {
if (tablePaths.isEmpty()) {
return;
}
+ enqueueRemoval(reader -> reader.removeTables(tablePaths));
+ }
+
+ void removePartitions(Set<PhysicalTablePath> partitionPaths) {
+ if (partitionPaths.isEmpty()) {
+ return;
+ }
+ enqueueRemoval(reader -> reader.removePartitions(partitionPaths));
Review Comment:
This removal task queues behind an active snapshot fetch, and `wakeUp()`
only wakes the log scanner. In a test that requested removal during the first
row, the fetcher still read all eight rows before finishing the split. Could
the snapshot loop check for cancellation between rows so a large expired
snapshot doesn’t hold up surviving partitions on the same reader?
--
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]
