[bvolpato]

SpannerChangeStreamOrderedWithinKeyGloballyIT set the flush timer for a key from the first record that arrived for that key. This assumes that the first record to arrive is the earliest, but Beam does not guarantee arrival order. When a record from a later batch arrives first, the timer is too late and one flush contains every batch. The test then fails because it expects one output per batch.

The logs of failed GCP IO precommit runs show this directly. In this run, the third batch started at 04:20:48.003 and the first at 04:20:25.714. Keys 1 and 2 then logged Setting timer at 2026-11-01T04:20:58.003Z, which is the third batch plus the 10 second interval. This run shows the same pattern.

This change moves the timer back when a record with an earlier commit timestamp arrives. A record that is not late holds the watermark back until it arrives, so the timer cannot fire too early. The test keeps its per-batch assertion.

SpannerChangeStreamOrderedByTimestampAndTransactionIdIT has similar timer logic, but its assertion does not depend on the number of flushes, so I did not change it.

To validate this without GCP, I replayed the commit timestamps from the failed run through the test's DoFns with TestStream on DirectRunner, in a scratch test that is not part of this PR. With the old logic, commit order passes and every out-of-order arrival tested fails with merged output like the CI failure. With this change, every arrival order tested produces the expected output. Spotless, Checkstyle, and test compilation ran locally. The integration test needs GCP, so it was not run locally. The GCP IO precommit on this PR runs it.

[github-actions]

📋 Review Summary

This PR fixes a flaky test failure in SpannerChangeStreamOrderedWithinKeyGloballyIT caused by assuming DataChangeRecord elements arrive at the stateful BufferKeyUntilOutputTimestamp transform in commit-timestamp order. By tracking the scheduled timer timestamp in a ValueState<Instant> and moving the timer earlier when a record with an earlier commit timestamp arrives, the per-key buffer reliably flushes per batch as intended.

🔍 General Feedback

• The root cause analysis and fix are sound: because non-late records hold back the event-time watermark, moving the event-time timer back when an earlier record arrives out of order is safe and prevents multiple batches from being merged into a single flush.
• State lifecycle management (timerTimestamp.write(...) on initial/subsequent scheduling and timerTimestamp.clear() when the buffer is empty) is handled cleanly.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[github-actions]

Assigning reviewers:

R: @kennknowles for label java.
Backup reviewer: ahmedabu98 (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).