Repository navigation
[Iceberg CDC sink] Commit files stage - #40134
Conversation
|
Assigning reviewers: R: @Abacn for label java. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
| * as its own single-snapshot commit. A configured {@code tokenHeartbeatMillis} adds an idle | ||
| * token-refresh timer ({@link #onHeartbeat}). | ||
| */ | ||
| static class OrderedCommitFn extends DoFn<KV<String, WindowedCommit>, KV<String, SnapshotInfo>> { |
There was a problem hiding this comment.
This file is extremely long. Can we move some of the inner classes and interfaces into separate files ? Also add dedicated test classes when appropriate.
There was a problem hiding this comment.
Did some refactoring. Pulled out OrderedCommitFn and commiter metrics logic into separate files.
| // The "commit" timer is always armed at the earliest pending window's end. | ||
| long earliestPendingWindowEnd = c.timestamp().getMillis(); | ||
| // Current input watermark tells us that all past pending windows are safe to commit. | ||
| long inputWatermark = timer.getCurrentRelativeTime().getMillis(); |
There was a problem hiding this comment.
This might be runner dependent. Let's adjust or document (if the correctness is preserved).
There was a problem hiding this comment.
It's a small optimization to commit multiple windows (backlog) in one fire, but yeah it's runner dependent. Probably not worth having divergent behavior so will just remove it
| .apply("KeyByDestination", WithKeys.of(ShardDeltaFiles::getTableIdentifierString)) | ||
| .setCoder(KvCoder.of(StringUtf8Coder.of(), ShardDeltaFiles.coder())) | ||
| // One element per (dest, window): every shard's output for the pair. | ||
| .apply("GatherShardsPerWindow", GroupByKey.create()) |
There was a problem hiding this comment.
Do we have to handle late data via after this GBK similar to CommitWindows ?
There was a problem hiding this comment.
We already handle late data upstream in CommitWindows, so we don't produce any late ShardDeltaFiles here.
Adds the commit stage. Driven by a stateful DoFn that commits each window's data and equality-delete files as one snapshot.
Includes the following in each snapshot summary for idempotency:
On retry or restart, recovers the token from the table's ancestry to skip already-committed windows, with a heartbeat that keeps the token fresh while the stream is idle.
Part of #39979
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes.See the Contributor Guide for more tips on how to make review process smoother.
To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md
GitHub Actions Tests Status (on master branch)
See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.