wanglijie95 commented on code in PR #21743:
URL: https://github.com/apache/flink/pull/21743#discussion_r1089668469
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/InputsLocationsRetriever.java:
##########
@@ -18,30 +18,40 @@
package org.apache.flink.runtime.scheduler;
-import org.apache.flink.runtime.executiongraph.Execution;
+import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
import java.util.Collection;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-/** Component to retrieve the inputs locations of a {@link Execution}. */
+/** Component to retrieve the inputs locations of an {@link ExecutionVertex}.
*/
public interface InputsLocationsRetriever {
/**
- * Get the producers of the result partitions consumed by an execution.
+ * Get the consumed result partition groups of an execution vertex.
*
- * @param executionVertexId identifies the execution
- * @return the producers of the result partitions group by job vertex id
+ * @param executionVertexId identifies the execution vertex
+ * @return the consumed result partition groups
*/
- Collection<Collection<ExecutionVertexID>>
getConsumedResultPartitionsProducers(
+ Collection<ConsumedPartitionGroup> getConsumedPartitionGroups(
ExecutionVertexID executionVertexId);
/**
- * Get the task manager location future for an execution.
+ * Get the producer execution vertices of a consumed result partition
group.
*
- * @param executionVertexId identifying the execution
+ * @param consumedPartitionGroup the consumed result partition group
+ * @return the ids of producer execution vertices
+ */
+ Collection<ExecutionVertexID> getProducersOrConsumedPartitionGroup(
Review Comment:
Maybe `getProducersOrConsumedPartitionGroup` ->
`getProducersOfConsumedPartitionGroup`
##########
flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java:
##########
@@ -124,6 +125,15 @@ void enableCheckpointing(
Map<IntermediateDataSetID, IntermediateResult> getAllIntermediateResults();
+ /**
+ * Get an intermediate result partition by the given or throw an excetion
if the partition is
Review Comment:
excetion -> exception
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/ConsumedPartitionGroup.java:
##########
@@ -130,4 +138,12 @@ public boolean areAllPartitionsFinished() {
public ResultPartitionType getResultPartitionType() {
return resultPartitionType;
}
+
+ public ConsumerVertexGroup getConsumerVertexGroup() {
+ return checkNotNull(consumerVertexGroup, "ConsumerVertexGroup is not
properly set.");
+ }
+
+ public void setConsumerVertexGroup(ConsumerVertexGroup
consumerVertexGroup) {
Review Comment:
Maybe add check `this.consumerVertexGroup == null ||
this.consumerVertexGroup == consumerVertexGroup` here
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/ConsumerVertexGroup.java:
##########
@@ -66,4 +75,12 @@ public boolean isEmpty() {
public ExecutionVertexID getFirst() {
return iterator().next();
}
+
+ public ConsumedPartitionGroup getConsumedPartitionGroup() {
+ return checkNotNull(consumedPartitionGroup, "ConsumedPartitionGroup is
not properly set.");
+ }
+
+ public void setConsumedPartitionGroup(ConsumedPartitionGroup
consumedPartitionGroup) {
Review Comment:
Maybe add check `this.consumedPartitionGroup == null ||
this.consumedPartitionGroup == consumedPartitionGroup` here
##########
flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/ExecutionGraphToInputsLocationsRetrieverAdapterTest.java:
##########
@@ -24,104 +24,113 @@
import org.apache.flink.runtime.executiongraph.ExecutionVertex;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSet;
+import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobmaster.TestingLogicalSlot;
import org.apache.flink.runtime.jobmaster.TestingLogicalSlotBuilder;
+import org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
import org.apache.flink.testutils.TestingUtils;
-import org.apache.flink.testutils.executor.TestExecutorResource;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.testutils.executor.TestExecutorExtension;
+import org.apache.flink.util.IterableUtils;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.util.Collection;
-import java.util.Collections;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.stream.Collectors;
-import static org.hamcrest.Matchers.empty;
-import static org.hamcrest.Matchers.hasItem;
-import static org.hamcrest.Matchers.hasSize;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link ExecutionGraphToInputsLocationsRetrieverAdapter}. */
-public class ExecutionGraphToInputsLocationsRetrieverAdapterTest extends
TestLogger {
+class ExecutionGraphToInputsLocationsRetrieverAdapterTest {
- @ClassRule
- public static final TestExecutorResource<ScheduledExecutorService>
EXECUTOR_RESOURCE =
- TestingUtils.defaultExecutorResource();
+ @RegisterExtension
+ static final TestExecutorExtension<ScheduledExecutorService>
EXECUTOR_EXTENSION =
+ TestingUtils.defaultExecutorExtension();
- /** Tests that can get the producers of consumed result partitions. */
@Test
- public void testGetConsumedResultPartitionsProducers() throws Exception {
+ void testGetConsumedPartitionGroupsAndProducers() throws Exception {
Review Comment:
`getProducersOfConsumedPartitionGroup` was not tested in this test
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultPreferredLocationsRetriever.java:
##########
@@ -84,11 +86,24 @@ private CompletableFuture<Collection<TaskManagerLocation>>
getPreferredLocations
CompletableFuture<Collection<TaskManagerLocation>> preferredLocations =
CompletableFuture.completedFuture(Collections.emptyList());
- final Collection<Collection<ExecutionVertexID>> allProducers =
-
inputsLocationsRetriever.getConsumedResultPartitionsProducers(executionVertexId);
- for (Collection<ExecutionVertexID> producers : allProducers) {
+ final Collection<ConsumedPartitionGroup> consumedPartitionGroups =
+
inputsLocationsRetriever.getConsumedPartitionGroups(executionVertexId);
+ for (ConsumedPartitionGroup consumedPartitionGroup :
consumedPartitionGroups) {
+ // Ignore the location of a consumed partition group if it has too
many distinct
+ // consumers compared to the consumed partition group size. This
is to avoid tasks
+ // unevenly distributed on nodes when running batch jobs or
running jobs in
+ // session/standalone mode.
+ if ((double) consumedPartitionGroup.getConsumerVertexGroup().size()
+ / consumedPartitionGroup.size()
+ > MAX_DISTINCT_LOCATIONS_TO_CONSIDER) {
Review Comment:
Maybe introduce a new static field `MAX_DISTINCT_CONSUMERS_TO_CONSIDER` ?
The `MAX_DISTINCT_LOCATIONS_TO_CONSIDER` does not quite match the meaning here.
--
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]
