[Flink Runner] Add new Source classes that are based on FLIP-27 Source API. #25525

becketqin · 2023-02-17T06:01:47Z

This is the first patch of migrating Flink runner batch job execution from DataSet to DataStream API.

The patch does the following:

Add two FLIP-27 source implementations: FlinkBoundedSource, FlinkUnboundedSource. ImpulseSource was implemented as a built-in type of FlinkBoundedSource.
Introduce necessary compatibility bridging classes to make the code work for Flink version [1.12, 1.15].

CHANGES.md was not modified with this patch because this patch does not change the existing execution paths. The followup patches will update CHANGES.md.

One notable difference between FLIP-27 Source implementations and the existing UnboundedSourceWrapper class is that the new FLIP-27 source does not support accumulators yet. So the beam metrics from the MetricsContainer will not be reported. However, the FLIP-27 based sources will emit their own metrics.

fixes #25486

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

becketqin · 2023-02-17T06:04:27Z

@xinyuiscool would you have time to take a look? Thanks!

becketqin · 2023-02-17T16:02:39Z

The failed PreCommit test is irrelevant to this patch.

github-actions · 2023-02-18T13:34:02Z

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

xinyuiscool · 2023-02-22T21:54:06Z

Run Java PreCommit

xinyuiscool

Reviewed half part. Will continue later.

...java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/source/FlinkSource.java

xinyuiscool · 2023-02-24T18:16:03Z

...ers/flink/translation/wrappers/streaming/io/source/unbounded/FlinkUnboundedSourceReader.java

+        newFuture.complete(null);
+      } else {
+        LOG.debug("There is no data available, scheduling the idle reader checker.");
+        scheduleTask(


This part is a bit unclear to me. Seems this checker thread will complete the dataAvailableFuture without start() being called. Does it mean that the reader will start to poll? Or isAvailable() will be invoked again?

start() is called only once right after the instantiation of the reader object. After that, the reader main thread will block on the future returned from isAvailable() if there is no data available for read. Before the main thread goes to block on that future, it sets the check thread to wake itself up after SLEEP_ON_IDLE_MS by completing that future. So the reader main thread will start to poll() again to check if there is more data available.

Logically speaking this is as if the reader main thread calls Thread.sleep(SLEEP_ON_IDLE_MS), except that we don't want to hijack the reader main thread in this case, because the reader main thread is also the task main thread and may need to do other things (e.g. checkpointing) even if there is no data available for reading.

Thanks for the explanation. So here is the intention is to put a wait for the next poll. I was thinking somewhere dataAvailableFuture will be set once the data is available. Looks like it's not used that way.

xinyuiscool · 2023-02-24T18:22:30Z

...pache/beam/runners/flink/translation/wrappers/streaming/io/source/FlinkSourceReaderBase.java

+    checkExceptionAndMaybeThrow();
+    LOG.info("Adding splits {}", splits);
+    sourceSplits.addAll(splits);
+    waitingForSplitChangeFuture.complete(null);


If notifyNoMoreSplits() is bound to be called after all splits are added, we only need to complete this future in the notifyNoMoreSplit(), right? it's a bit weird that we complete this future in two places and then in line 174 we recreate the future.

If we only need to complete this future once, then we can mark the var final so it's easier to understand.

In general, there is no guarantee that notifyNoMoreSplits is always going to be invoked, or when it will be invoked.

I think we need to complete the future in both addSplits() and notifyNoMoreSplits(). Basically if a reader goes to sleep because it has exhausted all the splits, it needs to be waken up either when a new split is assigned or NoMoreSplits notification is received.

Our current SplitEnumerator implementation follows a static splits assignment approach, so it sends all the splits to a subtask at once and then sends the NoMoreSplits notification immediately. So timing wise, it seems that the reader can wait for the NoMoreSplits notification and then act. However, that won't work for the dynamic assignment case. For example, if a reader only gets one split at a time and will request another split from the SplitEnumerator after finishing reading from the current split, it has to wake up and poll once a split is assigned. So the splitChangeFuture has to be completed in addSplits(). Also, if the reader has exhausted all the splits and gone to sleep, it has to be waken up upon receiving NoMoreSplits notification so it can exit normally.

Orthogonally, it seems the code does have a bug here. If there are live readers, the future returned by isAvailableForAliveReaders() should also be completed when there is a split change, either from addSplits() or notifyNoMoreSplits(). I'll update the patch to fix that.

xinyuiscool · 2023-02-24T18:30:15Z

...pache/beam/runners/flink/translation/wrappers/streaming/io/source/FlinkSourceReaderBase.java

+  // ------------------------------ private methods ------------------------------
+
+  @SuppressWarnings("unchecked")
+  private <CheckpointMarkT extends UnboundedSource.CheckpointMark>


This method uses the OutputT from the enclosing class. So it is not static. We can make it static, but the benefit might be limited given we don't expect tons of reader instances. And it also makes the code somewhat less readable.

xinyuiscool · 2023-02-24T18:30:33Z

...pache/beam/runners/flink/translation/wrappers/streaming/io/source/FlinkSourceReaderBase.java

+    }
+  }
+
+  private Source.Reader<T> createReader(@Nonnull FlinkSourceSplit<T> sourceSplit)


Similar to above, this method uses the type T from the enclosing class.

xinyuiscool · 2023-02-24T18:30:48Z

...pache/beam/runners/flink/translation/wrappers/streaming/io/source/FlinkSourceReaderBase.java

+    }
+  }
+
+  private <CheckpointMarkT extends UnboundedSource.CheckpointMark>


Same as above.

xinyuiscool · 2023-02-24T18:53:40Z

.../beam/runners/flink/translation/wrappers/streaming/io/source/FlinkSourceSplitEnumerator.java

+    while (splitIter.hasNext()) {
+      Map.Entry<Integer, List<FlinkSourceSplit<T>>> entry = splitIter.next();
+      int readerIndex = entry.getKey();
+      int targetSubtask = readerIndex % context.currentParallelism();


By reading the above code, seems the key is already the targetSubtask, so maybe we don't need to do this again?

Not sure if I follow the comment. The key will be in the pendingSplits map because that is an "intended assignments" regardless of whether the target reader has actually registered to the SplitEnumerator or not. But we can only send the "intended assignments" to a reader if that reader has registered to the SplitEnumerator. That is what the check does.

becketqin

@xinyuiscool Thanks for the comments. I'll update the patch.

becketqin · 2023-02-25T00:27:02Z

...pache/beam/runners/flink/translation/wrappers/streaming/io/source/FlinkSourceReaderBase.java

+    checkExceptionAndMaybeThrow();
+    LOG.info("Adding splits {}", splits);
+    sourceSplits.addAll(splits);
+    waitingForSplitChangeFuture.complete(null);


In general, there is no guarantee that notifyNoMoreSplits is always going to be invoked, or when it will be invoked.

I think we need to complete the future in both addSplits() and notifyNoMoreSplits(). Basically if a reader goes to sleep because it has exhausted all the splits, it needs to be waken up either when a new split is assigned or NoMoreSplits notification is received.

Our current SplitEnumerator implementation follows a static splits assignment approach, so it sends all the splits to a subtask at once and then sends the NoMoreSplits notification immediately. So timing wise, it seems that the reader can wait for the NoMoreSplits notification and then act. However, that won't work for the dynamic assignment case. For example, if a reader only gets one split at a time and will request another split from the SplitEnumerator after finishing reading from the current split, it has to wake up and poll once a split is assigned. So the splitChangeFuture has to be completed in addSplits(). Also, if the reader has exhausted all the splits and gone to sleep, it has to be waken up upon receiving NoMoreSplits notification so it can exit normally.

Orthogonally, it seems the code does have a bug here. If there are live readers, the future returned by isAvailableForAliveReaders() should also be completed when there is a split change, either from addSplits() or notifyNoMoreSplits(). I'll update the patch to fix that.

becketqin · 2023-02-25T00:33:16Z

...pache/beam/runners/flink/translation/wrappers/streaming/io/source/FlinkSourceReaderBase.java

+  // ------------------------------ private methods ------------------------------
+
+  @SuppressWarnings("unchecked")
+  private <CheckpointMarkT extends UnboundedSource.CheckpointMark>


This method uses the OutputT from the enclosing class. So it is not static. We can make it static, but the benefit might be limited given we don't expect tons of reader instances. And it also makes the code somewhat less readable.

becketqin · 2023-02-25T00:36:26Z

...pache/beam/runners/flink/translation/wrappers/streaming/io/source/FlinkSourceReaderBase.java

+    }
+  }
+
+  private Source.Reader<T> createReader(@Nonnull FlinkSourceSplit<T> sourceSplit)


Similar to above, this method uses the type T from the enclosing class.

becketqin · 2023-02-25T00:37:30Z

...pache/beam/runners/flink/translation/wrappers/streaming/io/source/FlinkSourceReaderBase.java

+    }
+  }
+
+  private <CheckpointMarkT extends UnboundedSource.CheckpointMark>


Same as above.

becketqin · 2023-02-25T00:42:01Z

.../beam/runners/flink/translation/wrappers/streaming/io/source/FlinkSourceSplitEnumerator.java

+    while (splitIter.hasNext()) {
+      Map.Entry<Integer, List<FlinkSourceSplit<T>>> entry = splitIter.next();
+      int readerIndex = entry.getKey();
+      int targetSubtask = readerIndex % context.currentParallelism();


Not sure if I follow the comment. The key will be in the pendingSplits map because that is an "intended assignments" regardless of whether the target reader has actually registered to the SplitEnumerator or not. But we can only send the "intended assignments" to a reader if that reader has registered to the SplitEnumerator. That is what the check does.

becketqin · 2023-02-25T00:53:17Z

...ers/flink/translation/wrappers/streaming/io/source/unbounded/FlinkUnboundedSourceReader.java

+        newFuture.complete(null);
+      } else {
+        LOG.debug("There is no data available, scheduling the idle reader checker.");
+        scheduleTask(


start() is called only once right after the instantiation of the reader object. After that, the reader main thread will block on the future returned from isAvailable() if there is no data available for read. Before the main thread goes to block on that future, it sets the check thread to wake itself up after SLEEP_ON_IDLE_MS by completing that future. So the reader main thread will start to poll() again to check if there is more data available.

Logically speaking this is as if the reader main thread calls Thread.sleep(SLEEP_ON_IDLE_MS), except that we don't want to hijack the reader main thread in this case, because the reader main thread is also the task main thread and may need to do other things (e.g. checkpointing) even if there is no data available for reading.

becketqin

@xinyuiscool Thanks for the comments. I have updated the patch to address them.

xinyuiscool

LGTM!

xinyuiscool · 2023-03-06T17:50:58Z

...ers/flink/translation/wrappers/streaming/io/source/unbounded/FlinkUnboundedSourceReader.java

+        newFuture.complete(null);
+      } else {
+        LOG.debug("There is no data available, scheduling the idle reader checker.");
+        scheduleTask(


Thanks for the explanation. So here is the intention is to put a wait for the next poll. I was thinking somewhere dataAvailableFuture will be set once the data is available. Looks like it's not used that way.

…e API. (apache#25525)

becketqin force-pushed the datastream-migration-1 branch from d572623 to ef85675 Compare February 17, 2023 08:21

xinyuiscool reviewed Feb 24, 2023

View reviewed changes

becketqin commented Feb 25, 2023

View reviewed changes

Jiangjie Qin added 2 commits February 25, 2023 10:03

Add new Source classes that are based on FLIP-27 Source API.

3516f4c

Address comments

1ba0471

becketqin force-pushed the datastream-migration-1 branch from ef85675 to 1ba0471 Compare February 25, 2023 02:04

github-actions bot added flink runners labels Feb 25, 2023

becketqin commented Feb 25, 2023

View reviewed changes

xinyuiscool approved these changes Mar 6, 2023

View reviewed changes

xinyuiscool merged commit 6452dc7 into apache:master Mar 6, 2023

becketqin mentioned this pull request Mar 7, 2023

[Task]: Add MetricsContainer support in the FLIP-27 sources in Flink runner. #25741

Closed

15 tasks

ruslan-ikhsan pushed a commit to akvelon/beam that referenced this pull request Mar 10, 2023

[Flink Runner] Add new Source classes that are based on FLIP-27 Sourc…

9abcf54

…e API. (apache#25525)

jto mentioned this pull request Sep 21, 2023

Add MetricsContainer support to the Flink sources. #25753

Closed

3 tasks

je-ik mentioned this pull request Nov 29, 2023

[Bug]: Impulse in FlinkRunner does not emit watermark #29558

Closed

16 tasks

kkdoon pushed a commit to twitter-forks/beam that referenced this pull request Feb 7, 2024

[Flink Runner] Add new Source classes that are based on FLIP-27 Sourc…

a6e4130

…e API. (apache#25525)

je-ik mentioned this pull request Apr 15, 2024

[Bug]: Messages are not ACK on Pubsub starting Beam 2.52.0 on Flink Runner in detached mode #29902

Closed

16 tasks

je-ik mentioned this pull request May 16, 2024

[Bug]: Watermarks and Windowing Not Working with FlinkRunner and KinesisIO Read Transform #31085

Closed

16 tasks

je-ik mentioned this pull request Jun 6, 2024

Backlog metrics do not showing up in FlinkRunner #29793

Closed

3 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Flink Runner] Add new Source classes that are based on FLIP-27 Source API. #25525

[Flink Runner] Add new Source classes that are based on FLIP-27 Source API. #25525

becketqin commented Feb 17, 2023 •

edited

Loading

becketqin commented Feb 17, 2023

becketqin commented Feb 17, 2023

github-actions bot commented Feb 18, 2023

xinyuiscool commented Feb 22, 2023

xinyuiscool left a comment

xinyuiscool Feb 24, 2023

becketqin Feb 25, 2023

xinyuiscool Mar 6, 2023

xinyuiscool Feb 24, 2023

becketqin Feb 25, 2023

xinyuiscool Feb 24, 2023

becketqin Feb 25, 2023

xinyuiscool Feb 24, 2023

becketqin Feb 25, 2023

xinyuiscool Feb 24, 2023

becketqin Feb 25, 2023

xinyuiscool Feb 24, 2023

becketqin Feb 25, 2023

becketqin left a comment

becketqin Feb 25, 2023

becketqin Feb 25, 2023

becketqin Feb 25, 2023

becketqin Feb 25, 2023

becketqin Feb 25, 2023

becketqin Feb 25, 2023

becketqin left a comment

xinyuiscool left a comment

xinyuiscool Mar 6, 2023

[Flink Runner] Add new Source classes that are based on FLIP-27 Source API. #25525

[Flink Runner] Add new Source classes that are based on FLIP-27 Source API. #25525

Conversation

becketqin commented Feb 17, 2023 • edited Loading

GitHub Actions Tests Status (on master branch)

becketqin commented Feb 17, 2023

becketqin commented Feb 17, 2023

github-actions bot commented Feb 18, 2023

xinyuiscool commented Feb 22, 2023

xinyuiscool 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

becketqin 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

becketqin left a comment

Choose a reason for hiding this comment

xinyuiscool left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

becketqin commented Feb 17, 2023 •

edited

Loading