jiangzzhu opened a new pull request, #40999: URL: https://github.com/apache/beam/pull/40999
## Summary Fixes sub-millisecond watermark truncation in `QueryChangeStreamAction` when persisting partition watermarks to the Spanner `PartitionMetadata` table. ### Problem - Spanner change stream records (`ChangeStreamRecord.getRecordTimestamp()`) and the `Watermark` column in the `PartitionMetadata` table use `com.google.cloud.Timestamp` (microsecond/nanosecond precision). - However, `QueryChangeStreamAction.updateWatermarkCallback` previously derived the persisted watermark from `WatermarkEstimator<org.joda.time.Instant>.currentWatermark()` via `Timestamp.ofTimeMicroseconds(watermark.getMillis() * 1_000L)`. - Because `org.joda.time.Instant` only has millisecond resolution, converting `Timestamp -> Instant -> Timestamp` truncated sub-millisecond precision (up to `999µs` / `999,999ns`), rewinding the persisted `Watermark` in `PartitionMetadata` to the millisecond floor of the last processed record's timestamp. - When a pipeline drains and resumes from `partition.getWatermark()`, `READ_CHANGE_STREAM` starts at the truncated millisecond floor instead of the last processed timestamp, rewinding the query by up to `999µs` and re-emitting already-processed records in that sub-millisecond window. Additionally, if a partition's `StartTimestamp` has sub-millisecond precision and records arrive within the same millisecond, the truncated watermark is `< StartTimestamp`, causing `PartitionMetadataDao.updateWatermark` (`if (partitionWatermark.compareTo(watermark) < 0)`) to skip advancing the watermark in Spanner. ### Solution - Expose `getLastClaimedPosition()` on `TimestampRangeTracker` (which already tracks the exact `com.google.cloud.Timestamp` on every successful `tryClaim`). - Track the last claimed full-precision `com.google.cloud.Timestamp` in `QueryChangeStreamAction.run` and pass it directly to `partitionMetadataDao.updateWatermark(composedToken, watermark)` in `updateWatermarkCallback`, without converting through `org.joda.time.Instant`. - Avoid registering `updateWatermarkCallback` when no record was claimed in the bundle. - Keep `ReadChangeStreamPartitionDoFn`'s `@WatermarkEstimatorState` as `org.joda.time.Instant` so Beam SDF watermark estimation and Dataflow `--update` state compatibility remain unchanged. ### Testing - Updated existing unit tests in `QueryChangeStreamActionTest` and added `testQueryChangeStreamPreservesSubMillisecondWatermarkPrecision` to verify full sub-millisecond `Timestamp` precision is preserved when updating `PartitionMetadata` watermarks. -- 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]
