[jiangzzhu]

Summary

Fixes partition stranding and SDF RestrictionTracker ordering in DetectNewPartitionsDoFn:

• Problem:
  1. DetectNewPartitionsAction.schedulePartitions previously called updateBatchToScheduled(batchPartitionsDifferentCreatedAt) (committing State = SCHEDULED in Spanner) before calling tracker.tryClaim(batchCreatedAt). If tryClaim failed, partitions were mutated to SCHEDULED without being emitted, violating the SDF RestrictionTracker contract.
  2. DetectNewPartitionsRangeTracker inherited non-checkpoint dynamic splitting (fractionOfRemainder != 0) from TimestampRangeTracker. Because DetectNewPartitionsDoFn is a singleton partition-discovery coordinator over [start, Timestamp.MAX_VALUE), a non-checkpoint split (fractionOfRemainder > 0) could both spawn an idle future restriction and split mid-batch (T1 < T_split <= Tk) after an earlier batch was claimed, allowing tryClaim(Tk) to fail and satisfy checkDone() while stranding T1 (< T_split) below the residual [T_split, MAX).
• Fix:
  • Move updateBatchToScheduled to after tracker.tryClaim(batchCreatedAt) succeeds in DetectNewPartitionsAction.schedulePartitions, immediately returning ProcessContinuation.stop() without performing Spanner writes or emitting outputs when tryClaim returns false, while preserving the 200-partition cross-timestamp batching.
  • Override DetectNewPartitionsRangeTracker.trySplit (matching Bigtable Change Streams' DetectNewPartitionsTracker) to refuse non-checkpoint splits (fractionOfRemainder != 0), while continuing to delegate checkpoints (fractionOfRemainder == 0) so every split boundary aligns with lastClaimedPosition + 1ns (or from if unclaimed) and never bisects an unclaimed batch.

Testing

• Updated DetectNewPartitionsActionTest to verify:
  • Cross-timestamp batching (< 200 partitions) claims the batch's latest CreatedAt timestamp before dao.updateToScheduled and outputs all partitions in the batch.
  • Returning ProcessContinuation.stop() without calling dao.updateToScheduled or receiver.outputWithTimestamp when tryClaim returns false (both for the first batch and for a subsequent batch after a full 200-partition batch).
• Added unit tests in DetectNewPartitionsRangeTrackerTest verifying trySplit returns null when fractionOfRemainder != 0 and checkpoints cleanly when fractionOfRemainder == 0.

[jiangzzhu]

@chenxuesdu @scwhittle could you please review this PR? Thanks!

[github-actions]

Assigning reviewers:

R: @nielm for label spanner.
Backup reviewer: chamikaramj (if the main reviewer is unresponsive, you can escalate to them by tagging their username)

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[scwhittle]

Actually, I'm not sure this would lead to lost things since if we haven't claimed the offset then on restart we would resume from the last claimed position.

[scwhittle]

I think that this breaks how restrictions are supposed to be used.
If a claim fails, none of the work associated after the previous successful claim should be performed. See the design doc: https://docs.google.com/document/d/1AQmx-T9XjSi1PNoEp5_L-lT0j7BkgTbmQnc6uFEMI4c/edit?tab=t.0#heading=h.3624h6581nxd

I think that the existing logic should:

1. should move the call to updateBatchToScheduled to be after tryClaim succeeds
2. Could possibly be changed to perform claims for each timestamp as it iterates over batches instead of the batches. And then once all timestamps are claimed, the batch gets too lare or upon claim failure the already claimed work could be updated to scheduled and output. That said, dataflow ensures that no splits are performed if there hasn't yet been a single claim, so it seems the existing logic would claim the entire first batch regardless and may make more progress if polling partitions are slow.

[jiangzzhu]

Thanks for catching this! Agreed — performing work or emitting outputs after tryClaim returns false violates the RestrictionTracker contract.

Updated per (1) to keep the existing 200-partition batching and move updateBatchToScheduled to after tryClaim(batchCreatedAt) succeeds. Also overrode DetectNewPartitionsRangeTracker.trySplit to return null when fractionOfRemainder != 0 (matching Bigtable Change Streams' DetectNewPartitionsTracker), so only checkpoints (fractionOfRemainder == 0) split the coordinator restriction and a split can never bisect an unclaimed cross-timestamp batch or spawn an idle future restriction. PTAL!

[jiangzzhu]

Actually, I'm not sure this would lead to lost things since if we haven't claimed the offset then on restart we would resume from the last claimed position.

@scwhittle That's true for checkpoints (trySplit(0)), where the residual starts at lastClaimedPosition + 1ns. The stranding case happens on a non-checkpoint dynamic split (trySplit(fraction > 0)) after an earlier batch was claimed: if splitPosition lands between T1 and T2 of a cross-timestamp batch (T1 < T_split <= T2), calling tryClaim(T2) fails while advancing lastAttemptedPosition past T_split (so checkDone() passes on primary [from, T_split) without outputting T1), and the residual [T_split, MAX) starts past T1.

Addressed both in the latest commit by moving updateBatchToScheduled after tryClaim and refusing non-checkpoint splits (fractionOfRemainder != 0) in DetectNewPartitionsRangeTracker — PTAL!

[scwhittle]

is it possible to remove this?

[jiangzzhu]

Done — removed @SuppressWarnings("nullness") and switched position.equals(lastAttemptedPosition) to Objects.equals(position, lastAttemptedPosition) so Checker Framework passes cleanly. Thanks for the review!