You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "tsreaper (via GitHub)" <gi...@apache.org> on 2024/07/08 07:41:41 UTC
[PR] [flink] Fix multi-add conflict when two batch jobs are writing into the same partition at the same time [paimon]
tsreaper opened a new pull request, #3699:
URL: https://github.com/apache/paimon/pull/3699
### Purpose
In new versions of Flink, if a batch job fails, it might restart from some operator in the middle.
If the job is restarted from the commit operator, endInput will be called again, and the same commit messages will be committed again.
So when `endInput` is called, we must check if these commit messages are already committed.
### Tests
* IT case.
### API and Format
No changes.
### Documentation
No new feature.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] Fix multi-add conflict when two batch jobs are writing into the same partition at the same time [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #3699:
URL: https://github.com/apache/paimon/pull/3699#discussion_r1668163529
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CommitterOperator.java:
##########
@@ -188,28 +189,37 @@ public void endInput() throws Exception {
}
pollInputs();
- commitUpToCheckpoint(Long.MAX_VALUE);
+ commitUpToCheckpoint(END_INPUT_CHECKPOINT_ID);
}
@Override
public void notifyCheckpointComplete(long checkpointId) throws Exception {
super.notifyCheckpointComplete(checkpointId);
- commitUpToCheckpoint(endInput ? Long.MAX_VALUE : checkpointId);
+ commitUpToCheckpoint(endInput ? END_INPUT_CHECKPOINT_ID : checkpointId);
}
private void commitUpToCheckpoint(long checkpointId) throws Exception {
NavigableMap<Long, GlobalCommitT> headMap =
committablesPerCheckpoint.headMap(checkpointId, true);
List<GlobalCommitT> committables = committables(headMap);
- committer.commit(committables);
- headMap.clear();
+ if (committables.isEmpty() && committer.forceCreatingSnapshot()) {
+ committables =
+ Collections.singletonList(
+ toCommittables(checkpointId, Collections.emptyList()));
+ }
- if (committables.isEmpty()) {
- if (committer.forceCreatingSnapshot()) {
- GlobalCommitT commit = toCommittables(checkpointId, Collections.emptyList());
- committer.commit(Collections.singletonList(commit));
- }
+ if (checkpointId == END_INPUT_CHECKPOINT_ID) {
+ // In new versions of Flink, if a batch job fails, it might restart from some operator
+ // in the middle.
+ // If the job is restarted from the commit operator, endInput will be called again, and
+ // the same commit messages will be committed again.
+ // So when `endInput` is called, we must check if these commit messages are already
+ // committed.
+ committer.filterAndCommit(committables);
Review Comment:
Here should filter commit, but should not check add files. The latter has a relatively high cost.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] Fix multi-add conflict when two batch jobs are writing into the same partition at the same time [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #3699:
URL: https://github.com/apache/paimon/pull/3699
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org