zhuzhurk commented on a change in pull request #15310:
URL: https://github.com/apache/flink/pull/15310#discussion_r599350956
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adapter/DefaultSchedulingPipelinedRegion.java
##########
@@ -62,21 +85,34 @@ public DefaultExecutionVertex getVertex(final
ExecutionVertexID vertexId) {
@Override
public Iterable<DefaultResultPartition> getConsumedResults() {
- if (consumedResults == null) {
+ if (consumedPartitionGroups == null) {
initializeConsumedResults();
}
- return consumedResults;
+ return () -> flatMap(consumedPartitionGroups,
resultPartitionRetriever);
}
private void initializeConsumedResults() {
- final Set<DefaultResultPartition> consumedResults = new HashSet<>();
+ final Set<ConsumedPartitionGroup> consumedResultGroupSet = new
HashSet<>();
for (DefaultExecutionVertex executionVertex :
executionVertices.values()) {
- for (DefaultResultPartition resultPartition :
executionVertex.getConsumedResults()) {
+ for (ConsumedPartitionGroup consumedResultGroup :
+ executionVertex.getConsumedPartitionGroups()) {
+ SchedulingResultPartition resultPartition =
+
resultPartitionRetriever.apply(consumedResultGroup.getFirst());
Review comment:
this is not correct because `consumedResultGroup` can be across multiple
regions. And it is also a problem if a `ConsumedPartitionGroup` contains both
intra-region consumed partition and inter-region one.
I think we need to have a clear idea to avoid this case or solve it before
we continue with this PR.
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
