jad-grepr opened a new pull request, #18374:
URL: https://github.com/apache/iceberg/pull/18374
## Problem
The Flink committers (`DynamicCommitter`, `IcebergFilesCommitter`,
`IcebergCommitter`) record a writer's last committed checkpoint only in the
snapshot summary, and they find it by walking the branch's ancestry from the
head. The walk stops at the first expired snapshot.
So when a writer is idle while other writers or table maintenance keep
committing, snapshot expiry eventually removes the chain between the head and
that writer's last commit, and the marker is lost. Restored from its savepoint,
the writer then tries to commit committables it already committed:
- **If the delta manifests were deleted** (normal after a commit), the job
fails reading them and restarts into the same failure.
- **If their deletion had failed**, the rows are duplicated.
## Change
**1. The writer's checkpoint in a table property.** Every commit stages the
snapshot update and
`flink.max-committed-checkpoint-id.<operator>.<job>.<branch>` (JSON: checkpoint
id, updated-at) in one `Transaction`, so both land or neither does. The summary
keys stay, and `flink.branch` is added.
**2. Lookups take the largest of the properties and the summary walk.**
Tables written before this change keep working. A checkpoint found only in the
summaries is recorded right after the lookup, under `<key>.<checkpoint>`. That
key is add-only, so a late or repeated request can never lower the value, on
any catalog.
**3. The walk is branch-scoped.** A branch that a commit forks from main's
head no longer inherits main's marker for the same checkpoint. Before, that
dropped the second branch's commit.
**4. The validator checks both.** `DynamicCommitter`'s validator reads the
transaction's properties, which every retry refreshes.
**5. The end-of-input checkpoint stays in the summary only, as before.**
**6. Retiring markers is explicit.**
`FlinkCommitMarkers.removeMarkers(Table, Predicate)` does it, re-selecting on
every retry. Committers never remove markers.
## Questions for reviewers
- **Opt-in?** Should the property markers sit behind a sink option? This
adds a persisted convention to table properties, visible in `SHOW
TBLPROPERTIES`.
- **Backports:** `v2.2` and `v1.20` would follow in a separate PR.
## Known limitations
- **Reused job ids (#18098) are unchanged.** Writers are identified by job
id, so a job id must not be reused by a run that starts without state.
- **`IcebergCommitter` still attributes a mixed-job batch to the last
request's job.** That's the static sink's analogue of #16011.
- **`removeMarkers` on REST catalogs.** REST applies property-only changes
without checking the base, so callers must only select retired writers.
- **Events at staging.** Inside a transaction, the snapshot producer emits
its events and commit report when the update is staged, before the publication.
## Tests
- **Idle writer past expiry** for all three committers, shown failing
without the change.
- **Recovery cases:**
- compaction, then expiry;
- a duplicate landing during the publication, with the re-apply and the
validator;
- unknown outcomes, published and not;
- a failed publication;
- racing writers;
- the same checkpoint on two branches.
- **Migration and downgrade:** summary-only markers, including with an empty
pending checkpoint.
- **A format-v2 batch** with an equality delete that fails partway and
resumes after its last committed checkpoint.
- **`TestFlinkCommitMarkers`:** the codec, canonical keys, the branch-scoped
walk, recording, and removal under races.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
