dmvk commented on code in PR #33606:
URL: https://github.com/apache/beam/pull/33606#discussion_r1920572946
##########
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:
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.
--
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]
