[FLINK-18934] Idle stream does not advance watermark in connected stream #15771

dawidwys · 2021-04-26T15:06:54Z

What is the purpose of the change

Make two and multi-input operators respect stream idleness.

Brief change log

extract a reusable CombinedWatermark class
drop StreamStatusMaintainer and StreamStatusProvider interfaces
Let the StreamStatus flow through the input/output chains

Two open questions:

do we want to maintain the behaviour from the removed tests? Those tests assumed that there can be records flowing when a stream is IDLE. In such cases we dropped any flowing Watermarks
Do we want to put the logic for making the stream temporary ACTIVE/IDLE in the RecordWriterOutput or do we want to find a better place for it.

Verifying this change

Adjusted tests that were verifying the idleness, e.g.

MultipleInputStreamTaskTest#testWatermarkAndStreamStatusForwarding
TwoInputStreamTaskTest#testWatermarkAndStreamStatusForwarding

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/Mesos, 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-04-26T15:10:23Z

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 77caa97 (Sat Aug 28 12:12:16 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.

Details

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

...ng-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperatorV2.java

flinkbot · 2021-04-26T15:39:55Z

CI report:

77caa97 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

...ng-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperatorV2.java

AHeise

Thank you very much for your contribution. The code definitively feels much cleaner with the additional abstraction of CombinedWatermark.

However, I'd probably not expose the inner class at all and get rid of the List ancestor. You now have a weird data structure that is unnecessarily mutable in many ways.

I'm also not sure if the tests are sufficient: I'm missing tests that explicitly test the behavior of 2- and N-ary operators. Most tests have been traditionally on task level but that is probably not the best place anymore.

Please add ticket and component to all commits (none of them are hotfixes).

flink-core/src/main/java/org/apache/flink/api/common/eventtime/CombinedWatermark.java

AHeise · 2021-05-11T09:02:16Z

...ming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java

+        combinedWatermark.add(new CombinedWatermark.PartialWatermark());
+        combinedWatermark.add(new CombinedWatermark.PartialWatermark());


shouldn't that depend on the number of inputs? Can't we create that in ctor?

I can add an additional ctor. (I was actually thinking about it) However, we still need the add method for the WatermarkOutputMultiplexer.

Hm, haven't thought about the dynamic cases (actually Kafka is not really dynamic but I can see that other sources are). Still the comment stands that I wouldn't expose PartialWatermark.

You could have a similar method to register in CombinedWatermark that returns the index of the newly added partial. The index is then used in private final Map<String, Integer> watermarkIndexPerOutputId; This index can then be used to update the CombinedWatermark via index.

This approach makes more sense if you can easily have all side-effects inside the CombinedWatermark#updateWatermark(int index, Watermark watermark). So it avoids the case, where you update the partial and then tell the combined watermark: "hey something changed, please check".

However, the current way may still be valid, if you explicitly want to have the information drift and only update at specific times for performance reasons. I'm currently not in the position to assess that properly.

Note that I don't like the current way because you don't have proper invariants (and I like having them) because of the information drift.

So there are two things here. My comment in this thread referred just to the add method, not exposing the PartialWatermark. Sure, if the issue was just about the add method, I would not expose the PartialWatermark.

The second problem is the combination of PartialWatermark and updateCombinedWatermark. Actually, you wrote the exact reason why we need it:

However, the current way may still be valid if you explicitly want to have the information drift and only update at specific times for performance reasons. I'm currently not in the position to assess that properly.

That's exactly what happens in the WatermarkOutputMultiplexer. It exposes two access patterns:

getImmediateOutput

getDeferredOutput

One immediately updates the combined watermark, and the other one does not. AFAIK the deferred one is used for periodic watermarks. Bear in mind that the PeriodicWatermark has the default scope. It is exposed only to the org.apache.flink.api.common.eventtime package for it to be used by WatermarkOutputMultiplexer.

Okay I'd still think that you could also hide that within #updateWatermark(..., boolean deferred) but the benefit of encapsulation gets even smaller (if it even still exists). So leave as is.

...ming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java

...ng-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperatorV2.java

...eaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSourceContexts.java

AHeise · 2021-05-11T09:19:45Z

...k-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/RecordWriterOutput.java


    private <X> void pushToRecordWriter(StreamRecord<X> record) {
-        serializationDelegate.setInstance(record);
+        if (announcedStatus.isIdle()) {


I think we need to add a comment here to explain why this can happen.

The comment is good in respect to explaining the invariant (you could also give a link to the docs). However, it's not helping the reader to understand why this case is happening.

...k-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/RecordWriterOutput.java

dawidwys · 2021-05-17T13:27:35Z

I tried addressing your comments. I am still not 100% sure about the short ACTIVE/IDLE cycle or should we rather let records be generated but halt watermarks forwarding.

Do you mind taking another look @AHeise ?

AHeise · 2021-05-20T11:31:35Z

...k-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/RecordWriterOutput.java

+        // StreamStatus.IDLE requires that no records nor watermarks travel through the branch
+        // in order to keep the older behaviour that records could've been generated down the
+        // pipeline even though the sources were idle we go through a short ACTIVE/IDLE loop
+        if (announcedStatus.isIdle()) {
+            writeStreamStatus(StreamStatus.ACTIVE);
+        }

+        serializationDelegate.setInstance(record);
        try {
            recordWriter.emit(serializationDelegate);
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        }
+
+        if (announcedStatus.isIdle()) {
+            writeStreamStatus(StreamStatus.IDLE);
+        }


Can we factor out some reusable pattern?

try (announcedStatus.ensureActive(this::writeStreamStatus)) { serializationDelegate.setInstance(record); try { recordWriter.emit(serializationDelegate); } catch (Exception e) { throw new RuntimeException(e.getMessage(), e); } }

Nice one! Will do.

AHeise · 2021-05-20T11:32:22Z

I tried addressing your comments. I am still not 100% sure about the short ACTIVE/IDLE cycle or should we rather let records be generated but halt watermarks forwarding.

Do you mind taking another look @AHeise ?

Looks already quite good. As discussed offline, I think that the ACTIVE/IDLE cycle is a good start and we should just optimize some operators to make them cycling in batches (asyncIO). To that end, could you try to factor out some small reusable pattern? I gave an idea on the respective lines.

Other comments:

I have not yet giving up on making PartialWatermark private though ;) Please check and refute my idea.
I'd squash the squash commit.

AHeise

LGTM now ;)

…ected stream Watermark in the two and multi input operators is computed in operators. So far operators were unaware of the StreamStatus, therefore even if a whole input was IDLE it could still block increasing the Watermark. This commit makes operators aware of the StreamStatus. The contract of the StreamStatus is that if a stream is IDLE it should not emit records nor watermarks.

…ider The StreamStatus traverses the whole DAG and it's state should be kept on the operator level. Given those assumptions the maintainer & provider are no longer necessary.

rmetzger added the review=description? label Apr 26, 2021

dawidwys commented Apr 26, 2021

View reviewed changes

...ng-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperatorV2.java Outdated Show resolved Hide resolved

dawidwys changed the title ~~WIP~~ [FLINK-18934] Idle stream does not advance watermark in connected stream Apr 26, 2021

rmetzger added component=API/DataStream component=Runtime/Task labels Apr 26, 2021

pnowojski reviewed Apr 27, 2021

View reviewed changes

...ng-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperatorV2.java Outdated Show resolved Hide resolved

dawidwys force-pushed the flink-18934 branch from b59abdb to a265c75 Compare April 28, 2021 07:07

dawidwys force-pushed the flink-18934 branch 4 times, most recently from 61748f5 to b979ed5 Compare May 10, 2021 16:04

dawidwys marked this pull request as ready for review May 10, 2021 16:36

dawidwys requested a review from AHeise May 10, 2021 16:36

AHeise requested changes May 11, 2021

View reviewed changes

dawidwys force-pushed the flink-18934 branch 2 times, most recently from 9aaebb2 to 37dcf10 Compare May 17, 2021 13:24

dawidwys force-pushed the flink-18934 branch from 37dcf10 to e4b89d7 Compare May 17, 2021 15:47

AHeise reviewed May 20, 2021

View reviewed changes

dawidwys force-pushed the flink-18934 branch from e4b89d7 to d55f891 Compare May 21, 2021 13:30

AHeise approved these changes May 21, 2021

View reviewed changes

dawidwys added 3 commits May 24, 2021 22:09

[FLINK-18934][core] Extract reusable CombinedWatermark class

33c25bd

[FLINK-18934][runtime] Drop StreamStatusMaintainer & StreamStatusProv…

77caa97

…ider The StreamStatus traverses the whole DAG and it's state should be kept on the operator level. Given those assumptions the maintainer & provider are no longer necessary.

dawidwys force-pushed the flink-18934 branch from d55f891 to 77caa97 Compare May 24, 2021 20:13

dawidwys merged commit ee9f9b2 into apache:master May 25, 2021

		combinedWatermark.add(new CombinedWatermark.PartialWatermark());
		combinedWatermark.add(new CombinedWatermark.PartialWatermark());

[FLINK-18934] Idle stream does not advance watermark in connected stream #15771

[FLINK-18934] Idle stream does not advance watermark in connected stream #15771

Uh oh!

Conversation

dawidwys commented Apr 26, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

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

Uh oh!

flinkbot commented Apr 26, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Automated Checks

Review Progress

Uh oh!

Uh oh!

flinkbot commented Apr 26, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Uh oh!

AHeise left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AHeise May 20, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

dawidwys commented May 17, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AHeise commented May 20, 2021

Uh oh!

AHeise left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

dawidwys commented Apr 26, 2021 •

edited

Loading

flinkbot commented Apr 26, 2021 •

edited

Loading

flinkbot commented Apr 26, 2021 •

edited

Loading

AHeise left a comment •

edited

Loading

AHeise May 20, 2021 •

edited

Loading