[FLINK-24530][datastream] GlobalCommitter might not commit all records on drain #17536

AHeise · 2021-10-21T11:40:35Z

What is the purpose of the change

This PR refactors SinkOperator setup and ensures that GlobalCommitter does not use notifyCheckpointComplete anymore since it may actually invoked before all Committers are notified. Thus, the global committer receives an incomplete set of Committables which will cause incorrect results in a final checkpoint setting.

Brief change log

Refactor SinkOperator and CommitterOperator setup such that the actual distribution logic resides in SinkTransformationTranslator.
Change parallelism of Committer to p and ensure a blocking pipeline in batch mode.
Trigger GlobalCommitter when all downstream Committers emitted the committables.

Verifying this change

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

Added integration tests for end-to-end deployment with large payloads (100MB)
Extended integration test for recovery after master (JobManager) failure
Added test that validates that TaskInfo is transferred only once across recoveries
Manually verified the change by running a 4 node cluser with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

flinkbot · 2021-10-21T11:44:06Z

CI report:

b9629ca Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

flinkbot · 2021-10-21T11:45:31Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit a629d69 (Thu Oct 21 11:45:31 UTC 2021)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

…tructor into factory method. The nature of this constructor is transformative and should be a factory method accordingly to differentiate between primary and secondary construction.

…or improved commit performance. * Splits batch committer from global commmitter. * Ensures blocking exchange between writer and committer in batch mode. * Simplify CommitterHandler signatures. Note that the construction of the (global) committer handler will be overhauled in a future commit. The implementation in this commit is used for easier review and successful CI.

…bal committer. Before this commit, all committables where immediately forwarded to the global committer. Retried committables where not emitted at all since they have already been sent. This commit: * CommitterHandler's only emit successful committables. All non-failed committables are deemed successful. Note that this change assumes that failed committables are a subset of the input committables and no new instances are created to reflect failures. JavaDoc is adjusted accordingly. * Allows CommitterHandlers to return which committables have been successfully retried. * CommitterRetrier can send successfully retried elements downstream with a callback.

…tirely into SinkTransformationTranslator. Before this commit SinkOperatorFactory and CommitterOperatorFactory had knowledge about the execution mode and created the operator accordingly. With this commit: * SinkTransformationTranslator is the only place where knowledge about the execution mode exists. The translator directly chooses the appropriate CommitterHandler and passes that information to the operator factories. * Factories are now much simplified. The SinkOperatorFactory retained the logic of choosing the appropriate writer state handler as that is independent of the execution mode. * Since operator factories are serializable, CommitterHandler received a serializable Factory layer, such that we do not need to make the CommitterHandler serializable. This refactoring is a preparation for cases where the sink pipeline will become more complex in the future as now SinkTransformationTranslator is the only place that needs to be touched.

fapaul

The refactoring looks great! I left some comments regarding the cleanup

Additional I was wondering whether we also need a blocking exchange between the committer and globalCommitter.

fapaul · 2021-10-25T08:30:02Z

...rc/main/java/org/apache/flink/streaming/runtime/operators/sink/ForwardCommittingHandler.java

 */
-class ForwardCommittingHandler<CommT> extends AbstractCommitterHandler<CommT, CommT, CommT> {
+class ForwardCommittingHandler<CommT> extends AbstractCommitterHandler<CommT, Void> {


I am not sure whether the abstraction of ForwardingCommittingHandler and NoopCommittingHandler really makes sense anymore after the refactoring. In the end, it could be a simple boolean flag whether the committables should be sent downstream if there is a global committer.

They are still needed in the current SinkOperator:

In batch with committer, we need ForwardingCommittingHandler.

In stream/batch without any committer, we need NoopCommittingHandler.

We could replace them by booleans but it's getting a bit ugly:

In streaming, we only emit on notifyCheckpointCompleted.

In batch, we only emit on preSnapshotBarrier.
So you'd need two booleans afaik. It's certainly less code but I'm not sure if it's easier to understand.

fapaul · 2021-10-25T08:35:06Z

...c/main/java/org/apache/flink/streaming/runtime/translators/SinkTransformationTranslator.java

@@ -51,6 +53,7 @@
                Object, SinkTransformation<InputT, CommT, WriterStateT, GlobalCommT>> {

    protected static final Logger LOG = LoggerFactory.getLogger(SinkTransformationTranslator.class);
+    public static final TypeInformation<byte[]> BYTES = TypeInformation.of(byte[].class);


fapaul · 2021-10-25T08:43:05Z

...rc/main/java/org/apache/flink/streaming/runtime/operators/sink/AbstractCommitterHandler.java

    }

-    protected abstract void retry(List<StateT> recoveredCommittables)
-            throws IOException, InterruptedException;
+    protected Collection<CommT> retry(List<StateT> recoveredCommittables)


AFAICT StreamingCommitterHandler and BatchCommitterHandler override this method but have identical implementations

@Override protected Collection<CommT> retry(List<CommT> recoveredCommittables) throws IOException, InterruptedException { return commitAndReturnSuccess(recoveredCommittables); }

Why can't we move the implementation to this class in the retry method?

That's unfortunately not that easy because of the global committers: Currently all committers are emitting CommT and not GlobalCommT anymore after this refactor. This is possible because in fact the global committers are not emitting anything.
Now commitAndReturnSuccess is working on the internal type (GlobalCommT in case of global committers). Hence, the signature is conflicting here.
We could create mix-ins interfaces for non-global and global committers where we can implement them. The question is if that's simpler. We could also re-introduce an emit type to CommitterHandler.

fapaul · 2021-10-25T08:49:05Z

...ming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitRetrier.java

-        this(processingTimeService, committerHandler, SystemClock.getInstance());
+            ProcessingTimeService processingTimeService,
+            CommitterHandler<CommT> committerHandler,
+            ThrowingConsumer<? super Collection<CommT>, IOException> committableConsumer) {


I don't like passing a lambda here because it is only used for emitting. I think we can simplify it by only passing a boolean to determine if emitting is necessary.

Also currently the CommitterOperator#emitCommittables references the commiterRetrier I think this may lead to weird situations.

I used a lambda here as a callback for timer-based retry. Not sure how this can be solved differently.

I certainly would try to avoid having the retrier in the emitCommittables - that can probably lead to nasty stack exceptions.

fapaul · 2021-10-25T09:11:13Z

...g-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.java

+        default <T> T checkSerializerPresent(Optional<T> optional, boolean global) {
+            String scope = global ? " global" : "";
+            checkState(
+                    optional.isPresent(),
+                    "Internal error: a%s committer should only be created if the sink has a%s committable serializer.",
+                    scope,
+                    scope);
+            return optional.get();
+        }
+
+        default <T> T checkCommitterPresent(Optional<T> optional, boolean global) {
+            String scope = global ? " global" : "";
+            checkState(
+                    optional.isPresent(),
+                    "Expected a%s committer because%s committable serializer is set.",
+                    scope,
+                    scope);
+            return optional.get();
+        }
+    }


Not sure I really like this approach of sharing utility functions but having them in separate class probably does not change much.

How about I change the Factory interface to an abstract class? It might be weird because they are default methods?

The abstract class sounds better then the scope is also more limited and the methods are not exposed to external calleers.

rmetzger added component=API/DataStream component=Connectors/Common labels Oct 21, 2021

AHeise added 4 commits October 21, 2021 20:44

[FLINK-24530][datastream] Turn secondary StreamingCommitterState cons…

b718389

…tructor into factory method. The nature of this constructor is transformative and should be a factory method accordingly to differentiate between primary and secondary construction.

fapaul reviewed Oct 25, 2021

View reviewed changes

AHeise added 4 commits October 27, 2021 20:25

a

8cdefe6

b

b497327

c

918ac46

WIP

b9629ca

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-24530][datastream] GlobalCommitter might not commit all records on drain #17536

[FLINK-24530][datastream] GlobalCommitter might not commit all records on drain #17536

AHeise commented Oct 21, 2021

flinkbot commented Oct 21, 2021 •

edited

flinkbot commented Oct 21, 2021

fapaul left a comment

fapaul Oct 25, 2021

AHeise Oct 25, 2021

fapaul Oct 25, 2021

fapaul Oct 25, 2021

AHeise Oct 25, 2021

fapaul Oct 25, 2021

AHeise Oct 25, 2021

fapaul Oct 25, 2021

AHeise Oct 25, 2021

fapaul Oct 27, 2021

[FLINK-24530][datastream] GlobalCommitter might not commit all records on drain #17536

Are you sure you want to change the base?

[FLINK-24530][datastream] GlobalCommitter might not commit all records on drain #17536

Conversation

AHeise commented Oct 21, 2021

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Oct 21, 2021 • edited

CI report:

flinkbot commented Oct 21, 2021

Automated Checks

Review Progress

fapaul left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

flinkbot commented Oct 21, 2021 •

edited