[flink] FlinkRunner initializes the same split twice (#31313) #33606

je-ik · 2025-01-15T14:49:56Z

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 or the workflows README to see a list of phrases to trigger workflows.

github-actions · 2025-01-15T15:35:40Z

Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @johnjcasey added as fallback since no labels match configuration

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

je-ik · 2025-01-16T14:05:32Z

R: @dmvk

github-actions · 2025-01-16T14:06:48Z

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

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

dmvk

The fix makes sense, good catch 👍 My main question is whether not re-listing partitions breaks the contract or not 🤔

dmvk · 2025-01-16T14:11:15Z

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

-    this.splitsInitialized = false;
+    this.splitsInitialized = splitsInitialized;
+
+    LOG.info(


This happens only when the job starts, I think info is fine here.

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

dmvk · 2025-01-16T14:19:07Z

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

-    List<FlinkSourceSplit<T>> splitsForSubtask =
-        pendingSplits.computeIfAbsent(subtaskId, ignored -> new ArrayList<>());
-    splitsForSubtask.addAll(splits);
+    // reshuffle splits, needed after rescaling


👍 would it make sense to provide additional reasoning?

This change could be reverted. I kept it, because it might make a difference under one improbable situation - reader returns splits to enumerator, and just after that a rescale happens. In all other cases this is equivalent to the previous implementation.

Can you please revert it then?

I could, but why? It helps under a rare circumstances and does no harm otherwise, so ... why revert?

That's where I was headed. If it fixes a corner case it CAN'T be reverted and we should document reasoning about the corner case to provide context for whoever touches the code next.

I see. The code was originally to solve the rescaling issue, but the whole process works a little differently and the corner case that this might solve is highly speculative, so I reverted the code to the previous version.

dmvk · 2025-01-16T14:21:12Z

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

  public void start() {
-    context.callAsync(
-        () -> {
-          try {
-            LOG.info("Starting source {}", beamSource);
-            List<? extends Source<T>> beamSplitSourceList = splitBeamSource();
-            Map<Integer, List<FlinkSourceSplit<T>>> flinkSourceSplitsList = new HashMap<>();
-            int i = 0;
-            for (Source<T> beamSplitSource : beamSplitSourceList) {
-              int targetSubtask = i % context.currentParallelism();
-              List<FlinkSourceSplit<T>> splitsForTask =
-                  flinkSourceSplitsList.computeIfAbsent(
-                      targetSubtask, ignored -> new ArrayList<>());
-              splitsForTask.add(new FlinkSourceSplit<>(i, beamSplitSource));
-              i++;
+    if (!splitsInitialized) {


WDYT about minimizing the changset and minizing nesting by doing

@Override public void start() { if (!splitsInitialized) { initializeSplits() } }

?

hmm, is this correct? are splits guaranteed to be stable over time? is it possible that you for example want to re-list kafka partitions? How are new partitions handled in Beam?

This codepath works for deprecated Read transforms. Beam does modern IO using SDF, which can handle dynamic splitting. In Read, splitting is essentially static.

dmvk · 2025-01-16T14:22:26Z

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

@@ -117,7 +103,8 @@ public Boundedness getBoundedness() {

  @Override
  public SplitEnumerator<FlinkSourceSplit<T>, Map<Integer, List<FlinkSourceSplit<T>>>>
-      createEnumerator(SplitEnumeratorContext<FlinkSourceSplit<T>> enumContext) throws Exception {
+      createEnumerator(SplitEnumeratorContext<FlinkSourceSplit<T>> enumContext) {


nit: unrelated change -> separate commit

je-ik · 2025-01-16T15:16:54Z

The fix makes sense, good catch 👍 My main question is whether not re-listing partitions breaks the contract or not 🤔

No, under the legary Source Read API, splitting the source is a one-shot operation.

dmvk

LGTM 👍

codecov · 2025-01-20T13:30:27Z

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 57.38%. Comparing base (e1245d9) to head (960fe79).
Report is 211 commits behind head on master.

Additional details and impacted files

@@            Coverage Diff            @@
##             master   #33606   +/-   ##
=========================================
  Coverage     57.38%   57.38%           
  Complexity     1474     1474           
=========================================
  Files           973      973           
  Lines        154624   154624           
  Branches       1076     1076           
=========================================
  Hits          88733    88733           
  Misses        63689    63689           
  Partials       2202     2202

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[flink] FlinkRunner initializes the same split twice (apache#31313)

7680207

github-actions bot added runners flink labels Jan 15, 2025

je-ik mentioned this pull request Jan 15, 2025

[Bug]: KinesisIO source on FlinkRunner initializes the same splits twice #31313

Closed

16 tasks

github-actions bot added Next Action: Reviewers build and removed build labels Jan 15, 2025

je-ik added 2 commits January 16, 2025 14:58

[flink] Redistribute splits after rescaling (apache#31313)

b4c6271

Run validates runner

e86f6fc

je-ik force-pushed the 31313-multiple-splits branch from 142d3ba to e86f6fc Compare January 16, 2025 13:59

github-actions bot added build and removed build labels Jan 16, 2025

dmvk reviewed Jan 16, 2025

View reviewed changes

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

dmvk reviewed Jan 16, 2025

View reviewed changes

github-actions bot added the build label Jan 16, 2025

je-ik force-pushed the 31313-multiple-splits branch 2 times, most recently from 925b074 to 22db506 Compare January 16, 2025 15:15

je-ik force-pushed the 31313-multiple-splits branch from 22db506 to d2e1af5 Compare January 17, 2025 20:24

Review

960fe79

je-ik force-pushed the 31313-multiple-splits branch from d2e1af5 to 960fe79 Compare January 19, 2025 10:43

dmvk approved these changes Jan 20, 2025

View reviewed changes

je-ik merged commit f357174 into apache:master Jan 20, 2025
22 checks passed

je-ik deleted the 31313-multiple-splits branch January 20, 2025 13:26

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[flink] FlinkRunner initializes the same split twice (#31313) #33606

[flink] FlinkRunner initializes the same split twice (#31313) #33606

je-ik commented Jan 15, 2025

github-actions bot commented Jan 15, 2025

je-ik commented Jan 16, 2025

github-actions bot commented Jan 16, 2025

dmvk left a comment

dmvk Jan 16, 2025

je-ik Jan 16, 2025

dmvk Jan 16, 2025

je-ik Jan 16, 2025

dmvk Jan 17, 2025

je-ik Jan 17, 2025

dmvk Jan 17, 2025 •

edited

Loading

je-ik Jan 17, 2025

dmvk Jan 16, 2025

dmvk Jan 16, 2025

je-ik Jan 16, 2025

dmvk Jan 16, 2025

je-ik commented Jan 16, 2025

dmvk left a comment

codecov bot commented Jan 20, 2025

[flink] FlinkRunner initializes the same split twice (#31313) #33606

[flink] FlinkRunner initializes the same split twice (#31313) #33606

Conversation

je-ik commented Jan 15, 2025

GitHub Actions Tests Status (on master branch)

github-actions bot commented Jan 15, 2025

je-ik commented Jan 16, 2025

github-actions bot commented Jan 16, 2025

dmvk 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

dmvk Jan 17, 2025 • edited Loading

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

je-ik commented Jan 16, 2025

dmvk left a comment

Choose a reason for hiding this comment

codecov bot commented Jan 20, 2025

Codecov Report

dmvk Jan 17, 2025 •

edited

Loading