[FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits #20485

smattheis · 2022-08-07T18:27:56Z

What is the purpose of the change

The change implements FLIP-217 (https://cwiki.apache.org/confluence/display/FLINK/FLIP-217+Support+watermark+alignment+of+source+splits) to support watermark alignment of source splits.

Brief change log

Revised SplitFetcher threading model and add support to pause/resume SplitFetcher
Adds support for watermark alignment of individual source splits controlled by SourceOperator
Adds support for split watermark alignment for Kafka and Pulsar sources
Adds configuration parameter to allow unaligned splits as migration plan to support legacy sources that lack support for split alignment

Verifying this change

This change added and extended tests and can be verified as follows:

SplitFetcherTest extended to test pause/resume ofSplitFetcher
SourceOperatorSplitWatermarkAlignmentTest added to test split alignment based on split reader watermarks
SplitFetcherPauseResumeSplitReaderITCase added to test pause/resume of individual split readers

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

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): yes (according to FLIP-217)
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? docs (configuration parameter) and JavaDocs (changes of public APIs)

flinkbot · 2022-08-07T18:33:23Z

CI report:

991230a Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

mas-chen

I did a quick first pass and still need to look deeper into the SplitFetcher changes. Lots of code changes! Do we have performance tests with analysis on the split fetcher changes, even without split alignment enabled?

mas-chen · 2022-08-12T17:52:17Z

docs/layouts/shortcodes/generated/pipeline_configuration.html

            <td>Whether name of vertex includes topological index or not. When it is true, the name will have a prefix of index of the vertex, like '[vertex-0]Source: source'. It is false by default</td>
        </tr>
+        <tr>
+            <td><h5>pipeline.watermark-alignment.allow-unaligned-source-splits</h5></td>


I think allow-aligned-source-splits is more intuitive. The emphasis is on alignment, rather than unalignment like in checkpoints.

IMHO something like source.reader.synchronize.splits would be more understandable for users.

flink-connectors/flink-connector-base/pom.xml

...main/java/org/apache/flink/connector/base/source/reader/fetcher/PauseOrResumeSplitsTask.java

flink-core/src/main/java/org/apache/flink/api/connector/source/SourceReader.java

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

...java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java

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

mas-chen · 2022-08-16T07:03:34Z

...pache/flink/connector/base/source/reader/fetcher/SplitFetcherPauseResumeSplitReaderTest.java

+
+        public void addPrefilledSplitsSingleReader(int numSplits, int numRecords) {
+            sourceReader.addSplits(createPrefilledSplits(numSplits, numRecords));
+            sourceReader.notifyNoMoreSplits();


Is this a required assumption? We should also exercise the case where splits are dynamically discovered (e.g. Kafka source topic partition discovery).

This is not required. The signal simply allows for a clean shutdown of the source.

...se/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherTest.java

mxm

Thank you @smattheis @AHeise @dawidwys! The changes definitely fill a gap in the source framework. There are a lot of changes here which are not always directly related to FLIP-217 like the refactoring of the threading and locking logic. Ideally, this could have done more incremental but I understand that was difficult due to the multiple authors.

I've left some comments and suggestions. All in all, the changes look good.

I wonder about the way the SourceOperator calls through the stack via pauseOrResumeSplits while also being called back with watermarks. My intuition would have been to do the pausing directly at the splits based on the current max watermark lag, e.g. by putting splits into a suspended mode where no more output is accepted from a split. I guess this could be tricky because the implementation of the source doesn't necessarily know it's paused and could timeout or otherwise misbehave. So it looks like the current implementation is simpler.

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

...main/java/org/apache/flink/connector/base/source/reader/fetcher/PauseOrResumeSplitsTask.java

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

.../src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java

...n/java/org/apache/flink/connector/kafka/source/reader/fetcher/KafkaSourceFetcherManager.java

...va/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java

mxm · 2022-08-23T12:59:31Z

...va/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java

+            Collection<PulsarPartitionSplit> splitsToPause,
+            Collection<PulsarPartitionSplit> splitsToResume) {
+        if (splitsToPause.size() > 1 || splitsToResume.size() > 1) {
+            throw new IllegalStateException("This pulsar split reader only support one split.");


I don't understand why we can't pause single splits. Even if so, I don't know why that should leak to the Pulsar reader.

Good question...

Have you found an answer for that @mxm ?

Yes, Pulsar splits are always started in a separate fetcher thread. We still want to be able to pause these split threads. So the code makes sense.

...java/org/apache/flink/streaming/api/operators/SourceOperatorSplitWatermarkAlignmentTest.java

mxm · 2022-08-29T12:57:49Z

Any thoughts on how and when to proceed with this PR? We'd like to help to bring this over the finish line.

pnowojski · 2022-08-30T07:11:49Z

I wonder about the way the SourceOperator calls through the stack via pauseOrResumeSplits while also being called back with watermarks. My intuition would have been to do the pausing directly at the splits based on the current max watermark lag, e.g. by putting splits into a suspended mode where no more output is accepted from a split. I guess this could be tricky because the implementation of the source doesn't necessarily know it's paused and could timeout or otherwise misbehave. So it looks like the current implementation is simpler.

Currently Splits are just pojos, data carriers, without much logic. If I understand you correctly, what you are suggesting would require us to push some availability logic down to the splits and it would still require handling of this availability in the SourceReader - basically the same logic that we have in the current proposal. The difference would be only that SourceReader would be informed about (un)availability change of a split, from the split, instead of from the pauseOrResume call.

Given that most of the users are expected to just use SourceReaderBase, the current proposal gives them watermark alignment for free. While adding availability logic to the splits would increase exposure of this feature in the public API, without much benefit?

Any thoughts on how and when to proceed with this PR? We'd like to help to bring this over the finish line.

Indeed there is an issue now, that I think @smattheis doesn't work on Flink anymore, so he can not finish this PR. I will go through the PR myself and do a quick pass/respond to the inlined questions from you @mxm and @mas-chen , but it would be great if someone could take over applying the actual changes after the review. Would you be up to do that @mxm ?

...r-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcher.java

...src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManager.java

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

.../src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaPartitionSplitReader.java

pnowojski · 2022-08-30T08:14:59Z

...va/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java

+            Collection<PulsarPartitionSplit> splitsToPause,
+            Collection<PulsarPartitionSplit> splitsToResume) {
+        if (splitsToPause.size() > 1 || splitsToResume.size() > 1) {
+            throw new IllegalStateException("This pulsar split reader only support one split.");


Good question...

flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java

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

mxm

@pnowojski Thank you for the comments. If nobody of the authors currently has time to work on this, I can take over the PR for now. I'd be glad to get another review from @mas-chen and you once I've addressed the comments.

pnowojski

If nobody of the authors currently has time to work on this, I can take over the PR for now.

That would be great :)

flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java

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

flink-core/src/main/java/org/apache/flink/api/connector/source/SourceReader.java

...ase/src/main/java/org/apache/flink/connector/base/source/reader/splitreader/SplitReader.java

flink-core/src/main/java/org/apache/flink/configuration/PipelineOptions.java

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

pnowojski

Thanks for the update @mxm , I've left a couple of minor comments.

The next question would be how to squash the commits? If you want to preserve authorship of your changes @mxm, without overwriting authors of the base commits, we would need either a single fixup commit from you, all one your fixup per original commit.

flink-core/src/main/java/org/apache/flink/api/connector/source/SourceReader.java

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

mxm · 2022-09-08T09:29:53Z

The next question would be how to squash the commits? If you want to preserve authorship of your changes @mxm, without overwriting authors of the base commits, we would need either a single fixup commit from you, all one your fixup per original commit.

I'd prefer this to be a single suqashed commit with the use of Co-Authored-By. IMHO there is no benefit in merging the individual commits because they are very much tight to each other. This keeps the history clean while retaining authorship. What do you think?

pnowojski · 2022-09-08T11:44:24Z

I'd prefer this to be a single suqashed commit with the use of Co-Authored-By. IMHO there is no benefit in merging the individual commits because they are very much tight to each other. This keeps the history clean while retaining authorship. What do you think?

I wouldn't squash all commits into one. The original commits authored by Sebastian, Dawid and Arvid I think are nicely separated. But sure, squashing your fixups to those commits and co-authoring them is also perfectly fine.

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

mxm · 2022-09-08T13:45:33Z

I wouldn't squash all commits into one. The original commits authored by Sebastian, Dawid and Arvid I think are nicely separated. But sure, squashing your fixups to those commits and co-authoring them is also perfectly fine.

If the commits are nicely separated, why were they not handled in separate PRs? Anyhow, I'm ok with just putting my changes in a single commit on top of those.

mxm · 2022-09-08T14:52:12Z

@flinkbot run azure

pnowojski

We are often splitting larger features into smaller commits for the sake of easier reviewing. As long as commits are independently working, and coherent (one need to find a sensible division) that's fine.

https://flink.apache.org/contributing/code-style-and-quality-pull-requests.html#separate-refactoring-cleanup-and-independent-changes

It gets more complicated in those rare cases where there are various authors, especially if someone takes over an orphaned PR. 🤷

LGTM assuming green azure.

Thanks a lot @mxm for driving this forward! Much appreciated. Thank you @dawidwys, @AHeise , @smattheis for your contributions as well!

smattheis · 2022-09-08T16:23:23Z

@mxm Thanks a lot also from my side for you contributions. However, @pnowojski this PR was not orphaned! (I wrote to you that I'm on vacations.)

mxm · 2022-09-08T17:48:29Z

Thank you @pnowojski for your thoughtful comments and guidance. And of course thanks to the original authors @AHeise @dawidwys @smattheis.

Whoops @smattheis I had assumed you wouldn't continue on this PR but I think no harm has been done. Let me know if anything is open on your side. It still need to revise the code with respect to a test failure.

pnowojski · 2022-09-08T19:28:01Z

@mxm Thanks a lot also from my side for you contributions. However, @pnowojski this PR was not orphaned! (I wrote to you that I'm on vacations.)

Yes, sorry for the mental shortcut. There was a delay in the communication between first my vacations and then yours. Around the time @mxm was asking how to continue with this PR, you @smattheis were not responding, and I got a message from you that you can work on it only after @mxm has already volunteered to take it over. After that I didn't want to create more confusion. But sorry for my inaccurate statement that it was "orphaned" :)

smattheis · 2022-09-09T09:32:53Z

@mxm @pnowojski All good. Thanks for driving it.

mxm · 2022-09-09T13:17:17Z

(Rebased)

mxm · 2022-09-09T13:19:17Z

@flinkbot run azure

pnowojski · 2022-09-12T09:52:52Z

There are still some related test failures:

Documented option pipeline.watermark-alignment.allow-unaligned-source-splits does not exist.

and

Sep 09 15:10:03 [ERROR] Errors:
Sep 09 15:10:03 [ERROR] PulsarUnorderedSourceReaderTest.supportsPausingOrResumingSplits(PulsarSourceReaderBase, Boundedness, String)[1] » Timeout
Sep 09 15:10:03 [ERROR] PulsarUnorderedSourceReaderTest.supportsPausingOrResumingSplits(PulsarSourceReaderBase, Boundedness, String)[2] » Timeout

mxm · 2022-09-12T10:28:30Z

Thanks @pnowojski. Will fix those.

On another note, do you think it makes sense to add a configuration option to disable using split alignment entirely? I'm a bit worried this feature might cause issues with the already integrated Kafka/Pulsar sources that we are not foreseeing right now. The added configuration option only allows to disable split alignments for non-compatible connectors. We could add another option to disable it entirely to retain the old behavior (disabled by default).

mxm · 2022-09-12T10:58:57Z

There are still some related test failures:

Documented option pipeline.watermark-alignment.allow-unaligned-source-splits does not exist.

I had to remove the @Deprecated flag from the configuration option. Otherwise, it will be undocumented.

mxm · 2022-09-12T10:59:47Z

@flinkbot run azure

pnowojski · 2022-09-12T15:29:40Z

On another note, do you think it makes sense to add a configuration option to disable using split alignment entirely? I'm a bit worried this feature might cause issues with the already integrated Kafka/Pulsar sources that we are not foreseeing right now. The added configuration option only allows to disable split alignments for non-compatible connectors. We could add another option to disable it entirely to retain the old behavior (disabled by default).

Does it make sense to use and enable watermark alignment via DataStream API to only disable it via config value? Keep in mind that user has to manually enable via WatermarkStrategy#withWatermarkAlignment call [1].

[1] https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/event-time/generating_watermarks/#watermark-alignment-_beta_

Instead of mixing 4 different concepts of synchronization, this commit aims to have one lock for everything and go through with it.

When there is only one SplitFetcher per split, the split advanced too much, and aligned event time is enabled, then the complete SplitFetcher can be paused by SplitFetcherManager#alignSplits, such that temporarily no more records are read from the respective split.

…dual splits This provides support for pausing and resuming individual splits in SourceReaderBase. This is used to align individual splits based on emitted watermarks in SourceOperator.

…termark drift

…rmark drift

…g option

…g tests

mxm · 2022-09-13T11:26:56Z

Fair point, users may modify the job to disable alignment. It doesn't make sense to override the watermark alignment settings from the config. But the same argument applies to the other configuration option to disable split alignment for not-ready sources. I think it would make sense to provide a transitory option to disable split alignment altogether until this feature has proven to be stable.

mxm · 2022-09-13T12:03:39Z

@flinkbot run azure

mxm · 2022-09-13T15:23:53Z

Tests are passing now.

mxm · 2022-09-14T12:56:46Z

Thank you everyone!

dawidwys · 2022-09-14T13:12:31Z

Thank you for helping getting it over the line! @mxm

tweise · 2022-09-14T13:38:06Z

Exciting to see this make the finish line! This was a long awaited feature (we implemented alignment for the Kinesis source ~4 years ago). Let's see how it holds up in production use cases!

Thanks @mxm @pnowojski @smattheis @dawidwys, @AHeise for your contributions!

pnowojski · 2022-09-16T08:45:36Z

Yes, it took longer than it should have. Good that it's finally been merged 🎉

flinkbot added component=Connectors/Common component=Runtime/Checkpointing labels Aug 7, 2022

smattheis force-pushed the flip-217-split-wm-alignment branch 2 times, most recently from 308488c to 2a4ae3e Compare August 8, 2022 05:46

mas-chen reviewed Aug 16, 2022

View reviewed changes

...se/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherTest.java Outdated Show resolved Hide resolved

mxm reviewed Aug 23, 2022

View reviewed changes

pnowojski reviewed Aug 30, 2022

View reviewed changes

mxm reviewed Sep 1, 2022

View reviewed changes

pnowojski reviewed Sep 2, 2022

View reviewed changes

pnowojski reviewed Sep 6, 2022

View reviewed changes

mxm force-pushed the flip-217-split-wm-alignment branch from f9b2e3b to a9246bd Compare September 7, 2022 15:02

mxm reviewed Sep 7, 2022

View reviewed changes

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java Show resolved Hide resolved

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java Show resolved Hide resolved

pnowojski reviewed Sep 8, 2022

View reviewed changes

flink-core/src/main/java/org/apache/flink/api/connector/source/SourceReader.java Outdated Show resolved Hide resolved

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

pnowojski reviewed Sep 8, 2022

View reviewed changes

flink-core/src/main/java/org/apache/flink/api/common/eventtime/CombinedWatermarkStatus.java Outdated Show resolved Hide resolved

mxm force-pushed the flip-217-split-wm-alignment branch from cd9ce51 to c2408e3 Compare September 8, 2022 13:51

pnowojski approved these changes Sep 8, 2022

View reviewed changes

mxm force-pushed the flip-217-split-wm-alignment branch 2 times, most recently from ff462cd to 22db9b0 Compare September 9, 2022 13:17

mxm force-pushed the flip-217-split-wm-alignment branch from 22db9b0 to d6aed81 Compare September 12, 2022 10:58

mxm force-pushed the flip-217-split-wm-alignment branch from d6aed81 to 546fc79 Compare September 12, 2022 10:59

Arvid Heise and others added 8 commits September 13, 2022 13:00

[FLINK-28853][connector-base] Revise threading model of SplitFetcher.

3b4df5d

Instead of mixing 4 different concepts of synchronization, this commit aims to have one lock for everything and go through with it.

[FLINK-28853][source][connector-base] Add support for aligning indivi…

fafe9af

…dual splits This provides support for pausing and resuming individual splits in SourceReaderBase. This is used to align individual splits based on emitted watermarks in SourceOperator.

[FLINK-28853][kafka] Implement pausing Kafka partitions in case of wa…

dd38cfb

…termark drift

[FLINK-28853][pulsar] Implement pausing Pulsar splits in case of wate…

7fa7da2

…rmark drift

[FLINK-28853][connector-base] Add allow-unaligned-source-splits confi…

520050a

…g option

[FLINK-28853][connector-base] Add source reader split alignment test

6e3ea66

[FLINK-28853] Address PR comments / Add Kafka and Pulsar split pausin…

991230a

…g tests

mxm force-pushed the flip-217-split-wm-alignment branch from 546fc79 to 991230a Compare September 13, 2022 11:21

mxm approved these changes Sep 13, 2022

View reviewed changes

mxm merged commit 0612a99 into apache:master Sep 14, 2022

[FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits #20485

[FLINK-28853][connector-base] Add FLIP-217 support for watermark alignment of source splits #20485

Uh oh!

Conversation

smattheis commented Aug 7, 2022

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 Aug 7, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

mas-chen left a comment

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!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

mxm left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

mxm commented Aug 29, 2022

Uh oh!

pnowojski commented Aug 30, 2022

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

mxm left a comment

Choose a reason for hiding this comment

Uh oh!

pnowojski left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

flinkbot commented Aug 7, 2022 •

edited

Loading

pnowojski left a comment •

edited

Loading

pnowojski left a comment •

edited

Loading

dawidwys commented Sep 14, 2022 •

edited

Loading