je-ik commented on code in PR #33606:
URL: https://github.com/apache/beam/pull/33606#discussion_r1920702553
##########
runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/source/FlinkSourceSplitEnumerator.java:
##########
@@ -111,10 +131,19 @@ public void handleSplitRequest(int subtaskId, @Nullable
String requesterHostname
@Override
public void addSplitsBack(List<FlinkSourceSplit<T>> splits, int subtaskId) {
- LOG.info("Adding splits {} back from subtask {}", splits, subtaskId);
- List<FlinkSourceSplit<T>> splitsForSubtask =
- pendingSplits.computeIfAbsent(subtaskId, ignored -> new ArrayList<>());
- splitsForSubtask.addAll(splits);
+ // reshuffle splits, needed after rescaling
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
