je-ik commented on code in PR #33606:
URL: https://github.com/apache/beam/pull/33606#discussion_r1920152547
##########
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 could, but why? It helps under a rare circumstances and does no harm
otherwise, so ... why revert?
--
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]
