You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2023/01/20 17:51:54 UTC
[GitHub] [flink] zhuzhurk opened a new pull request, #21743: [FLINK-15325][coordination] Ignores the input locations of a ConsumePartitionGroup if the corresponding ConsumerVertexGroup is too large
zhuzhurk opened a new pull request, #21743:
URL: https://github.com/apache/flink/pull/21743
## What is the purpose of the change
This change improves the `DefaultPreferredLocationsRetriever` so that it ignores the input locations of a ConsumePartitionGroup if the corresponding ConsumerVertexGroup is too large. This helps to avoid tasks to be unevenly distributed on nodes when running batch jobs or running jobs in session/standalone mode, because the consumers in this case will tend to be placed on the same node of the input vertex.
## Brief change log
*(for example:)*
- Changed `EdgeManagerBuildUtil` to set the ConsumedPartitionGroup/ConsumerVertexGroup to its corresponding ConsumerVertexGroup/ConsumedPartitionGroup
- Changed `DefaultPreferredLocationsRetriever` to ignore the input locations of a ConsumePartitionGroup if the corresponding ConsumerVertexGroup is too large(compared to the ConsumePartitionGroup)
## Verifying this change
- *Added unit tests in EdgeManagerBuildUtilTest and DefaultPreferredLocationsRetrieverTest*
- *The change is also covered by existing tests of preferred location retriever*
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (**yes** / no / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / **no**)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zhuzhurk commented on pull request #21743: [FLINK-15325][coordination] Ignores the input locations of a ConsumePartitionGroup if the corresponding ConsumerVertexGroup is too large
Posted by "zhuzhurk (via GitHub)" <gi...@apache.org>.
zhuzhurk commented on PR #21743:
URL: https://github.com/apache/flink/pull/21743#issuecomment-1408196997
Thanks for reviewing! @wanglijie95
Merging.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zhuzhurk closed pull request #21743: [FLINK-15325][coordination] Ignores the input locations of a ConsumePartitionGroup if the corresponding ConsumerVertexGroup is too large
Posted by "zhuzhurk (via GitHub)" <gi...@apache.org>.
zhuzhurk closed pull request #21743: [FLINK-15325][coordination] Ignores the input locations of a ConsumePartitionGroup if the corresponding ConsumerVertexGroup is too large
URL: https://github.com/apache/flink/pull/21743
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zhuzhurk commented on a diff in pull request #21743: [FLINK-15325][coordination] Ignores the input locations of a ConsumePartitionGroup if the corresponding ConsumerVertexGroup is too large
Posted by "zhuzhurk (via GitHub)" <gi...@apache.org>.
zhuzhurk commented on code in PR #21743:
URL: https://github.com/apache/flink/pull/21743#discussion_r1089677545
##########
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:
`this.consumedPartitionGroup == null` is good enough I think. We do not expect this set method of one group to be invoked multiple times.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #21743: [FLINK-15325][coordination] Ignores the input locations of a ConsumePartitionGroup if the corresponding ConsumerVertexGroup is too large
Posted by GitBox <gi...@apache.org>.
flinkbot commented on PR #21743:
URL: https://github.com/apache/flink/pull/21743#issuecomment-1398746472
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9118095bdfdf184466f83661e8b39661c7075181",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9118095bdfdf184466f83661e8b39661c7075181",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9118095bdfdf184466f83661e8b39661c7075181 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zhuzhurk commented on a diff in pull request #21743: [FLINK-15325][coordination] Ignores the input locations of a ConsumePartitionGroup if the corresponding ConsumerVertexGroup is too large
Posted by "zhuzhurk (via GitHub)" <gi...@apache.org>.
zhuzhurk commented on code in PR #21743:
URL: https://github.com/apache/flink/pull/21743#discussion_r1089678609
##########
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:
Good catch!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] wanglijie95 commented on a diff in pull request #21743: [FLINK-15325][coordination] Ignores the input locations of a ConsumePartitionGroup if the corresponding ConsumerVertexGroup is too large
Posted by "wanglijie95 (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zhuzhurk commented on a diff in pull request #21743: [FLINK-15325][coordination] Ignores the input locations of a ConsumePartitionGroup if the corresponding ConsumerVertexGroup is too large
Posted by "zhuzhurk (via GitHub)" <gi...@apache.org>.
zhuzhurk commented on code in PR #21743:
URL: https://github.com/apache/flink/pull/21743#discussion_r1089677576
##########
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:
`this. consumerVertexGroup == null` is good enough I think. We do not expect this set method of one group to be invoked multiple times.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org