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()`. The
regression covers 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.
##########
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:
Table-removal targets now also include the pending request paths, even when
there are no active splits for that table. This ensures both emitter state
cleanup and fetcher-side `removeTables()` still run, closing the cached `Table`
and clearing its metadata caches.
Added a reader regression where the last local split has already finished,
plus an SDK integration test that drops/recreates the table while another table
remains readable.
##########
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:
I used a fresh metadata check on the fetcher for this case, rather than
triggering a new Enumerator discovery. On `PartitionNotExistException`, it
calls `Admin.listPartitionInfos()` with a bounded wait and finishes the failed
split only if the response confirms that the partition is missing. If the
partition still exists, the original exception propagates; lookup
failures/timeouts also propagate. Other subscription exceptions are not
suppressed. The remaining live splits in the assignment continue, and hybrid
splits are queued for snapshot reading only after log subscription succeeds.
This keeps confirmation local without adding a fetcher-to-Enumerator failure
event and a coordinator round trip. Normal Enumerator discovery still performs
global removal coordination through request/ACK. The tradeoff is that local
split completion can precede that discovery; with periodic discovery disabled,
Enumerator reconciliation waits for the next startup discovery.
Confirmed-missing results are reused only within the current assignment batch,
and the next batch checks again.
Added SDK coverage for mixed/all-expired assignments and checkpoint restore
before/after finished-split processing, plus log/hybrid unit coverage. Would
this local confirmation approach address your concern?
##########
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:
Added table/partition cancellation markers that the reader thread publishes
before queuing cleanup. The snapshot loop checks the current split's marker
before polling a batch and between rows, so a matching snapshot yields to the
queued removal task. Scanner close, SDK unsubscribe, and finished-split
reporting remain on the fetcher thread.
The four table/partition × current/unrelated snapshot regressions verify
that removal after row 1 stops the affected eight-row batch at one row, while
an unrelated snapshot still reads all eight. They also check that scanner close
runs on the fetching thread. A `pollBatch()` already in progress is still
governed by its existing timeout; this does not make that SDK call immediately
interruptible.
--
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]
