fxbing commented on PR #4527: URL: https://github.com/apache/flink-cdc/pull/4527#issuecomment-5788185233
> ## Review: [[FLINK-40575](https://issues.apache.org/jira/browse/FLINK-40575)][pipeline-connector/fluss] Support dynamic table unsubscription and re-subscription > ### Strengths > * **Clean protocol design with authoritative semantics.** `TableSubscriptionEvent` (enumerator → reader, an authoritative full subscription snapshot with `subscribedTablePaths` / `pendingRemovalRequests` / `fencedTablePaths`) plus `TableRemovalAckEvent` (reader → enumerator, per-requestId acknowledgement) form a complete removal state machine. `requestId` is monotonically generated via `NEXT_REMOVAL_REQUEST_ID` and regenerated on restore / reader restart, naturally defending against stale acks (covered by `FlussSourceEnumeratorTest#testRestoreRegeneratesRemovalRequestIdAndRejectsOldAcknowledgement`). > * **The reader-side staging mechanism elegantly eliminates the event/split ordering race.** `FlussSourceReader.addSplits` buffers splits into `stagedSplits` until the first subscription snapshot arrives, and `snapshotState` includes staged splits in the checkpoint, so no split is lost. When the first snapshot arrives, splits are filtered by `isActive()` + `fencedTablePaths` before activation, so restored splits cannot resurrect an already-unsubscribed table. > * **The fence mechanism handles the checkpoint window boundary.** After tombstone cleanup, `removalFences` is set to -1, bound to a real checkpointId on the next `snapshotState`, and released only after `notifyCheckpointComplete`. `addSplitsBack` uses `fencedFreshSplits` to distinguish "splits assigned during the fence window" from stale splits. Combined with tests like `testRemovalFenceSurvivesAbortedCheckpointAndRearm` (fence re-arms after an aborted checkpoint), the cross-checkpoint consistency argument is complete. > * **Thread confinement is correct.** `FlussSourceFetcherManager.removeTables` wraps cleanup as a `SplitFetcherTask` submitted to the fetcher thread, avoiding cross-thread mutation of `FlussSplitReader` state; all new enumerator state changes happen on the coordinator thread (in `callAsync` callbacks), needing no locking. `currentLogScanner` is now `volatile`, and `wakeUp()` changed from a no-op to actually waking the scanner. > * **The resource cleanup chain is complete.** `FlussSplitReader.removeTables` unsubscribes log buckets, clears bounded splits, closes `Table` handles, and clears rowType/PK/partition caches; `FlussRecordEmitter.removeTable` uses `inactiveTablePaths` to block emission of in-flight enqueued records, and the new `FlussDeserializer.removeState` clears the 5 per-table caches in `FlussRecordDeserializer`; on re-subscribe, `applySplit` clears the inactive flag — a closed loop. > * **State serialization is backward-compatible.** `FlussSourceEnumStateSerializer` bumps VERSION 1→2, v1 state deserializes correctly (pending removal treated as empty set), unknown versions throw `IOException`, and all three scenarios have unit tests. > * **Test coverage is high quality for a PR of this size.** EnumeratorTest (+815 lines) covers late-init not resurrecting removed tables, mixed init callbacks, partition drops not producing table-level tombstones, and reader-restart waiting for a new snapshot; ITCase (+431 lines) covers savepoint restore not resurrecting unsubscribed tables, re-add going through a fresh earliest lifecycle, and remove+re-add within the same checkpoint window rolling back to contiguous semantics — exactly the scenarios most prone to error in this feature. > * **Docs (en/zh) accurately describe the boundary semantics**: discovery-result authority, empty result = full unsubscribe, discovery failure fails the job, restored reader waits for a snapshot, PK-table lease not released early, and the rollback semantics across checkpoint windows — all consistent with the code. > > ### Issues > #### Critical (must fix) > * None. I traced the core subscription lifecycle state machine, concurrency race protection, and checkpoint consistency path by path (including the timing where the coordinator checkpoint precedes the task barrier) and found no defect that would lose data, resurrect a removed table, or deadlock. > > #### Important (should fix) > * **[FlussSplitReader.java, `removeTables()`, ~L446-453] A `table.close()` failure throwing `IOException` directly fails the whole job.** The removal task runs on the fetcher thread, so an exception there goes down the fetcher-failure path and triggers job failover. But a Fluss `Table.close()` failure is very likely just a transient RPC issue, and at that point the split is already marked removed and the table is logically unsubscribed — failing the job as the cost of a cleanup failure is disproportionate (compare the existing convention in the same file's `close()`, where a `table.close()` failure only does `LOG.warn`). Suggestion: catch and `LOG.warn`, then continue cleaning up the remaining tables (best-effort), or at least ensure a failure on one table does not abort cleanup of the rest. > * **[FlussSourceEnumerator.java, `handleTableBucketChanges()`, ~L638-651] An initialization failure is silently swallowed with no logging.** When `error != null` and all initializing paths are already unsubscribed / in pending removal, the code returns directly — violating the basic "don't swallow exceptions" rule. In production, debugging "why does this batch of buckets have no splits" would leave no clue. Suggestion: even if you decide to ignore it, add `LOG.warn("Ignoring initialization failure for unsubscribed tables {}", initializingPaths, error)`. > * **[FlussSourceEnumerator.java, `removeConsumedKvSnapshotBuckets()` ~L933-947 + `checkTableBucketChanges` call order] There is an implicit timing coupling on `lastDiscoveredTableBuckets` being the _previous_ round's discovery result, and after restore the first discovery round has it empty.** `updateSubscriptions` runs before `lastDiscoveredTableBuckets = discoveryResult.tableBuckets` is assigned, so unsubscribe cleanup uses the previous round's bucket list — this order is correct but guaranteed only by code position, with no comment; if someone reorders the two lines, KV-snapshot consumed-record cleanup silently breaks. Also, in the first discovery round after a checkpoint restore, `lastDiscoveredTableBuckets` is empty; if an unsubscribe happens at that moment with an in-flight `FinishedKvSnapshotConsumeEvent`, `pendingRemovalBuckets` cannot be registered, and a removed table's bucket may be re-counted into `consumedKvSnapshotMap`. Suggestion: write the "read old value → u pdate to new value" contract into a comment or make it an explicit parameter; for the restore scenario, consider not handling removal until after the first discovery round following `start()`, or back-fill the registration using the current round's buckets. > > #### Minor (nice to have) > * **[FlussSourceEnumerator.java L103-105] `NEXT_REMOVAL_REQUEST_ID` is a static JVM-level counter, and the comment overstates it.** "unique across restored enumerator instances" only holds within the same JVM; across a JobManager process restart it recounts from 1 (harmless under the current design, since after restore the reader necessarily waits for a new snapshot and re-acks, but the comment should say so honestly, or switch to an instance field seeded with a timestamp/random value on restore). > * **[FlussSourceEnumerator.java, addSplitsBack fence filter] `split.equals(fencedFreshSplits.get(split.splitId()))` relies on value equality** (verified that `FlussLogSplit`/`FlussHybridSnapshotLogSplit` implement equals based on offset+tablePath+bucket): a stale split whose fields happen to equal a fresh split's would be wrongly admitted. Practically near-impossible (re-init goes through `scan.startup.mode` producing new offset semantics), but worth a comment noting this is intentional value-equality. > * **[FlussSourceEnumerator.java, updateSubscriptions] After `removeIf` on `pendingPartitionSplitAssignment`, empty Set entries may remain**, accumulating slightly over long runs; cleaning up empty entries would be nicer. > * **[FlussSplitReader.java, removeTables] `currentLogScanner.unsubscribe(...)` is not wrapped in try/catch.** If the Fluss client throws a RuntimeException, it likewise fails the job and the `tables`/cache cleanup won't run (partial cleanup state). Same root cause as the `close()` issue above — can be made best-effort together. > * **[FlussSourcePipelineITCase] Uses fixed `Thread.sleep(2s/4s/10s)` to wait for discovery cycles**, which risks flakiness on slow CI. Suggest polling for the condition instead (the file already has an `awaitEvents` helper; the sleeps are mainly for crossing the discovery interval — acceptable but improvable). > * **[TableRemovalAckEvent / TableSubscriptionEvent] Missing `toString()`.** SourceEvents print as object addresses in logs/debugging; a simple toString would greatly help when diagnosing distributed races. > > ### Verdict > **Ready to merge?** With fixes > > **Reasoning:** The subscription lifecycle state machine (tombstone → requestId ack → fence → checkpoint-bound release) is rigorously designed, staging + authoritative snapshots eliminate the event/split ordering race, and tests cover the vast majority of dangerous boundaries. However, two exception-handling issues on the cleanup path (`table.close()` failing the job, and initialization errors being silently swallowed) and the implicit timing coupling in `removeConsumedKvSnapshotBuckets` should be fixed before merging. @leonardBang thanks for your reviewing 1. **Table close failures:** I’ll log `Table.close()` failures and continue cleaning up the remaining tables and caches, with a regression test. Scanner unsubscription failures will still propagate, since unsubscription must succeed before removal can be acknowledged. 2. **Ignored initialization errors:** I’ll add a warning with the affected paths and exception. Errors will still propagate if any path in that initialization remains active. 3. **Previous-discovery dependency:** I’ll document why cleanup must use the previous discovery result. Restored splits remain staged until the first authoritative subscription snapshot and cannot report snapshot completion while staged, so additional recovery-time bucket tracking does not appear necessary. 4. **Request ID uniqueness:** I’ll clarify that uniqueness applies across enumerator instances within the same JVM and retain the existing counter. 5. **Value equality in the fence filter:** I suggest retaining this comparison. Equal splits currently have identical reader-visible behavior, while drop/recreate changes the table ID. The unequal stale/fresh case already has regression coverage. 6. **Empty assignment entries:** Their count is bounded by reader parallelism, and they do not affect assignment behavior, so I suggest leaving them unchanged. 7. **Scanner unsubscription failures:** I suggest preserving failure propagation. Continuing after an unsubscribe failure could acknowledge removal while the scanner still retains the old bucket subscription. 8. **Fixed waits in tests:** Negative assertions need a bounded observation window. I’ll strengthen the initial consumption waits with observable event conditions, so slow startup cannot silently weaken the savepoint assertions. 9. **Event `toString()`:** Agreed. I’ll add `toString()` implementations to both events to expose the subscription, removal request, and fence information during debugging. -- 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]
