Flink: Remove multi-commit-request state handling in DynamicCommitter - #18097
LouisDeconinck wants to merge 3 commits into
Conversation
DynamicCommitter kept a List of CommitRequests per checkpoint only to process Flink state from releases before 1.11, where the upstream DynamicWriteResultAggregator emitted multiple commit requests per checkpoint. This was scheduled for removal in 1.12; upgrading through 1.11 migrates the state to a single commit request per checkpoint. The committer now tracks a single CommitRequest per (table, branch, checkpoint) triplet and fails fast if a duplicate is received instead of silently dropping it. Generated-by: Devin
84551fd to
bc601a8
Compare
|
@aiborodin: Could you please review? |
| throws IOException { | ||
| long checkpointId = commitRequestMap.lastKey(); | ||
| List<ManifestFile> manifests = Lists.newArrayList(); | ||
| NavigableMap<Long, List<WriteResult>> pendingResults = Maps.newTreeMap(); |
There was a problem hiding this comment.
This should be simplified to NavigableMap<Long, WriteResult> and the loop below should use the WriteResult.Builder to merge multiple WriteResults from each FlinkManifestUtil.readCompletedFiles() call. Every method down the call stack should accept this simplified collection: replacePartitions, commitDeltaTxn.
There was a problem hiding this comment.
Done in dede42c — pending results are now NavigableMap<Long, WriteResult>, built per checkpoint via WriteResult.Builder; replacePartitions/commitDeltaTxn consume the merged results directly. Applied identically to the v1.20, v2.1, v2.2, and v2.3 implementations.
| .put(committable.checkpointId(), request); | ||
| Preconditions.checkState( | ||
| previous == null, | ||
| "Received multiple commit requests for table %s branch %s at checkpoint %s", |
There was a problem hiding this comment.
Jobs upgrading directly from 1.10 or earlier versions would hit this precondition and fail. If we are going down this path, let's make the error message more descriptive with actionable instructions for users on how to resolve this:
- Option 1: users must upgrade to 1.11 first. TLDR: The 1.11 committer commits or skips the old-format committables during initializeState and removes them from the collector, so any savepoint taken from the 1.11 job is already clean. One successful restore plus one savepoint is enough. Then upgrade to 1.12.
- Option 2 (unsure if this option would work across all Flink versions given FLINK-37605): stop-with-savepoint +
--drainflag and start job with no state.
We would also have to document these upgrade requirements somewhere, maybe in release notes of 1.12?
More generally, I think the cleanup is good but it comes with the cost of potentially creating some annoyance/loss of state for end users so I can also lean towards keeping this committable merging code for longer (given it's not much extra code). @pvary what's your opinion?
|
@aiborodin: Given that 1.12 is basically out, we can cheaply keep it a bit longer and remove it only to 1.13 release. |
…ades Generated-by: Devin Co-Authored-By: Louis Deconinck <louis.dck@gmail.com>
3 participants
Part of #16445
All APIs tagged
@deprecated ... will be removed in 1.12.0have already been removed on main. This PR removes the remaining code path documented for removal in 1.12:DynamicCommitter's per-checkpointList<CommitRequest>handling.DynamicCommittergrouped multipleCommitRequests per (table, branch, checkpoint) triplet only to process Flink state from releases before 1.11, where the upstreamDynamicWriteResultAggregatoremitted multiple committables per checkpoint. The code comment scheduled this for removal in 1.12, since upgrading through 1.11 migrates the state to a single commit request per checkpoint.Changes:
commit()now tracks a singleCommitRequest<DynamicCommittable>per checkpoint (matchingIcebergCommitter) and fails fast viaPreconditions.checkStateif a duplicate is received, instead of silently overwriting it and later signaling it as committed without its data files being committed.commitPendingRequestsiterates a single request per checkpoint directly.testTableBranchAtomicCommitForAppendOnlyDataandtestCommitDeltaTxnWithAppendFilesnow pack the multiple delta manifests into oneDynamicCommittable, preserving the same commit coverage.AI Disclosure