jiangzzhu opened a new pull request, #40977:
URL: https://github.com/apache/beam/pull/40977
## Summary
Adds diagnostic logging around Spanner Change Streams partition-discovery
termination, watermark cache refreshes, partition completion, and metadata
table cleanup to aid root-cause analysis if a streaming pipeline stops
unexpectedly:
- **`PartitionMetadataDao`**:
- Adds `getMetadataTableName()` and `getPartitionStateSummary()` (for both
GoogleSQL and PostgreSQL) returning per-`State` row counts, min/max
`Watermark`, and min/max `CreatedAt`.
- Logs a warning in `updateWatermark` if a non-monotonic watermark update
regressing by more than 1 second is skipped.
- **`AsyncWatermarkCache`**:
- Logs at `DEBUG` when advancing `lastCachedMinWatermark`.
- When `getUnfinishedMinWatermarkFrom(sinceTimestamp)` returns `null`,
logs `sinceTimestamp` and (when `sinceTimestamp > Timestamp.MIN_VALUE`) queries
`getUnfinishedMinWatermarkFrom(Timestamp.MIN_VALUE)` and
`getPartitionStateSummary()` to log a `WARN` if unfinished partitions still
exist with watermarks below `sinceTimestamp`.
- **`DetectNewPartitionsAction`**:
- Logs `INFO` when `tracker.tryClaim(batchCreatedAt)` returns `false` and
stops the current restriction.
- Enhances `terminate()` to include the metadata table name, current
restriction, and `getPartitionStateSummary()` (logging at `WARN` if
non-`FINISHED` rows still exist in the metadata table while preserving the
`"All partitions have been processed, stopping"` log message).
- **`CleanUpReadChangeStreamDoFn`**:
- Logs the metadata table name, indexes, and `getPartitionStateSummary()`
prior to dropping the metadata table (logging at `WARN` if unfinished
partitions remain).
- **`QueryChangeStreamAction`**:
- Logs `INFO` if `tracker.tryClaim(endTimestamp)` fails when finishing a
partition, and includes `startTimestamp`, `endTimestamp`, and current watermark
in the partition-finish log.
All steady-state polling paths remain free of extra `INFO`/`WARN` logs, and
diagnostic queries are wrapped in `try/catch` so they cannot disrupt normal
execution or cleanup.
## Testing
- Added unit tests in `PartitionMetadataDaoTest`,
`DetectNewPartitionsActionTest`, `AsyncWatermarkCacheTest`, and
`CleanUpReadChangeStreamDoFnTest`.
- Verified `:sdks:java:io:google-cloud-platform:spotlessApply`,
`compileJava` (Checker Framework), `checkstyleMain`, `checkstyleTest`, and
`:sdks:java:io:google-cloud-platform:test --tests
"org.apache.beam.sdk.io.gcp.spanner.changestreams.*"`.
--
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]
