[FLINK-36379] Improve (Global)Committer with UC disabled [1.20] #25660
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Use more of the assertj native patterns to compare results.
AHeise
force-pushed
the
FLINK-36379-1.20
branch
from
cfcf7a9
to
0b326b7
Compare
Global committers used to trail a full checkpoint behind committer. That means that data appeared only after >2*checkpoint interval in the sinks that use it (e.g. delta). However, committables in the global committers are already part of the first checkpoint and are idempotent: On recovery, they are resend from the committer to the global committer. Thus, the global committer can actually be seen as stateless and doesn't need to conduct its own 2PC protocol. This commit lets the global committer collect all committables on input (as before) but immediately tries to commit when it has received all (deducible from CommitterSummary - which was always the original intent of that message). Thus, in most cases, GlobalCommitter ignores notifyCheckpointCompleted now as the state of the checkpoint can be inferred by received committables from upstream. There are special cases where a global committer is directly chained to a writer. In this case, the global committer does need to conduct a 2PC protocol in place of the committer. To differentiate these cases, the global committer now has its own transformation. (cherry picked from commit 67be29a)
Without UCs, a committer doesn't need to do anything on #processInput except collecting. It emits only on notifyCheckpointCompleted (or endInput for batch). We can also harden some contracts: * NotifyCheckpointCompleted can assert that all committables are received. * Emit committables downstream only if all committables are finished. (cherry picked from commit 7f40ab9)
AHeise
force-pushed
the
FLINK-36379-1.20
branch
from
0b326b7
to
49d6a5a
Compare
|
Reviewed by Chi on 21/11/24. Looks in hand, test failures currently |
AHeise
force-pushed
the
FLINK-36379-1.20
branch
from
9652997
to
5b2ce39
Compare
davidradl
reviewed
Nov 22, 2024
| } | ||
|
|
||
| private static <T> void copySafely(Consumer<T> consumer, Supplier<T> provider) { | ||
| T value = provider.get(); |
There was a problem hiding this comment.
I am wondering
- are name, uid or uidHash nullable? The uid seems like a mandatory non nullable identifier.
- should we error if any of these values are null?
- or is there a valid case for having any of these values as null, if so should we not copy over the null?
davidradl
reviewed
Nov 22, 2024
| return translateInternal(transformation, false); | ||
| } | ||
|
|
||
| private Collection<Integer> translateInternal( |
There was a problem hiding this comment.
should we add junits for this method - or do you think the higher level tests already cover us?
davidradl
reviewed
Nov 22, 2024
| @@ -181,7 +181,7 @@ private void commitAndEmitCheckpoints() throws IOException, InterruptedException | |||
| private void commitAndEmit(CheckpointCommittableManager<CommT> committableManager) | |||
| throws IOException, InterruptedException { | |||
| Collection<CommittableWithLineage<CommT>> committed = committableManager.commit(committer); | |||
There was a problem hiding this comment.
NIT: as we are no longer using committed in the if, we could move committableManager.commit(committer) inlined to line 189 and remove the committed variable
davidradl
reviewed
Nov 22, 2024
| /** | ||
| * Commits all due committables if all respective committables of the specific subtask and | ||
| * checkpoint have been received. | ||
| * | ||
| * @param committer used to commit to the external system | ||
| * @return successfully committed committables with meta information | ||
| * @throws IOException |
There was a problem hiding this comment.
why are these removed, we still throw these exceptions.
davidradl
reviewed
Nov 22, 2024
| @@ -160,13 +169,37 @@ public Collection<CommittableWithLineage<CommT>> commit(Committer<CommT> committ | |||
| return committed; | |||
| } | |||
|
|
|||
| Collection<CommitRequestImpl<CommT>> getPendingRequests(boolean onlyIfFullyReceived) { | |||
| Collection<CommitRequestImpl<CommT>> getPendingRequests(boolean assertFull) { | |||
There was a problem hiding this comment.
could the junits cover all the permutations of this stream - so we know there are no side effects?
davidradl
reviewed
Nov 22, 2024
| if (transformation instanceof OneInputTransformation) { | ||
| StreamOperatorFactory<?> operatorFactory = | ||
| ((OneInputTransformation<?, ?>) transformation).getOperatorFactory(); | ||
| if (operatorFactory instanceof CommitterOperatorFactory) { |
There was a problem hiding this comment.
I think a comment to describe why we return true here would be useful.
davidradl
reviewed
Nov 22, 2024
| private <R> void repeatUntilConverged(Supplier<R> producer) { | ||
| R lastResult = producer.get(); | ||
| R nextResult; | ||
| while (!lastResult.equals(nextResult = producer.get())) { |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Backport of #25456 with some adjustments in
GlobalCommitterOperater(main doesn't haveGlobalCommittercode anymore).