You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/02/10 22:49:42 UTC
[1/3] flink git commit: [FLINK-7124] [flip6] Add test to verify
rescaling JobGraphs works correctly
Repository: flink
Updated Branches:
refs/heads/master 91eea376e -> 556ea8a71
[FLINK-7124] [flip6] Add test to verify rescaling JobGraphs works correctly
This commit adds two tests to verify behaviours of rescaling JobGraphs:
1. JobGraphs can be consecutively rescaled to arbitrary valid DOPs
2. Rescaling beyond max parallelism would fail
The second test, however, is temporarily disabled for now since it
doesn't properly fail.
This closes #4510.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/5fe31cbf
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/5fe31cbf
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/5fe31cbf
Branch: refs/heads/master
Commit: 5fe31cbf26f35ae6921d543d362d2d9f006f45db
Parents: 91eea37
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Authored: Thu Aug 10 13:41:40 2017 +0800
Committer: Till Rohrmann <tr...@apache.org>
Committed: Sat Feb 10 23:43:48 2018 +0100
----------------------------------------------------------------------
.../ExecutionGraphRescalingTest.java | 235 +++++++++++++++++++
.../executiongraph/ExecutionGraphTestUtils.java | 77 ++++++
2 files changed, 312 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/5fe31cbf/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRescalingTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRescalingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRescalingTest.java
new file mode 100644
index 0000000..e6cc908
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRescalingTest.java
@@ -0,0 +1,235 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.executiongraph;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.checkpoint.StandaloneCheckpointRecoveryFactory;
+import org.apache.flink.runtime.executiongraph.restart.NoRestartStrategy;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
+import org.apache.flink.runtime.jobgraph.DistributionPattern;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
+import org.apache.flink.runtime.jobmanager.scheduler.Scheduler;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
+
+import org.junit.Ignore;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+/**
+ * This class contains tests that verify when rescaling a {@link JobGraph},
+ * constructed {@link ExecutionGraph}s are correct.
+ */
+public class ExecutionGraphRescalingTest {
+
+ private static final Logger TEST_LOGGER = LoggerFactory.getLogger(ExecutionGraphRescalingTest.class);
+
+ @Test
+ public void testExecutionGraphArbitraryDopConstructionTest() throws Exception {
+
+ final Configuration config = new Configuration();
+
+ final JobVertex[] jobVertices = createVerticesForSimpleBipartiteJobGraph();
+ final JobGraph jobGraph = new JobGraph(jobVertices);
+
+ // TODO rescaling the JobGraph is currently only supported if the
+ // TODO configured parallelism is ExecutionConfig.PARALLELISM_AUTO_MAX.
+ // TODO this limitation should be removed.
+ for (JobVertex jv : jobVertices) {
+ jv.setParallelism(ExecutionConfig.PARALLELISM_AUTO_MAX);
+ }
+
+ ExecutionGraph eg = ExecutionGraphBuilder.buildGraph(
+ null,
+ jobGraph,
+ config,
+ TestingUtils.defaultExecutor(),
+ TestingUtils.defaultExecutor(),
+ new Scheduler(TestingUtils.defaultExecutionContext()),
+ Thread.currentThread().getContextClassLoader(),
+ new StandaloneCheckpointRecoveryFactory(),
+ AkkaUtils.getDefaultTimeout(),
+ new NoRestartStrategy(),
+ new UnregisteredMetricsGroup(),
+ 5,
+ TEST_LOGGER);
+
+ for (JobVertex jv : jobVertices) {
+ assertEquals(5, jv.getParallelism());
+ }
+ verifyGeneratedExecutionGraphOfSimpleBitartiteJobGraph(eg, jobVertices);
+
+ // --- verify scaling up works correctly ---
+
+ // TODO rescaling the JobGraph is currently only supported if the
+ // TODO configured parallelism is ExecutionConfig.PARALLELISM_AUTO_MAX.
+ // TODO this limitation should be removed.
+ for (JobVertex jv : jobVertices) {
+ jv.setParallelism(ExecutionConfig.PARALLELISM_AUTO_MAX);
+ }
+
+ eg = ExecutionGraphBuilder.buildGraph(
+ null,
+ jobGraph,
+ config,
+ TestingUtils.defaultExecutor(),
+ TestingUtils.defaultExecutor(),
+ new Scheduler(TestingUtils.defaultExecutionContext()),
+ Thread.currentThread().getContextClassLoader(),
+ new StandaloneCheckpointRecoveryFactory(),
+ AkkaUtils.getDefaultTimeout(),
+ new NoRestartStrategy(),
+ new UnregisteredMetricsGroup(),
+ 10,
+ TEST_LOGGER);
+
+ for (JobVertex jv : jobVertices) {
+ assertEquals(10, jv.getParallelism());
+ }
+ verifyGeneratedExecutionGraphOfSimpleBitartiteJobGraph(eg, jobVertices);
+
+ // --- verify down scaling works correctly ---
+
+ // TODO rescaling the JobGraph is currently only supported if the
+ // TODO configured parallelism is ExecutionConfig.PARALLELISM_AUTO_MAX.
+ // TODO this limitation should be removed.
+ for (JobVertex jv : jobVertices) {
+ jv.setParallelism(ExecutionConfig.PARALLELISM_AUTO_MAX);
+ }
+
+ eg = ExecutionGraphBuilder.buildGraph(
+ null,
+ jobGraph,
+ config,
+ TestingUtils.defaultExecutor(),
+ TestingUtils.defaultExecutor(),
+ new Scheduler(TestingUtils.defaultExecutionContext()),
+ Thread.currentThread().getContextClassLoader(),
+ new StandaloneCheckpointRecoveryFactory(),
+ AkkaUtils.getDefaultTimeout(),
+ new NoRestartStrategy(),
+ new UnregisteredMetricsGroup(),
+ 2,
+ TEST_LOGGER);
+
+ for (JobVertex jv : jobVertices) {
+ assertEquals(2, jv.getParallelism());
+ }
+ verifyGeneratedExecutionGraphOfSimpleBitartiteJobGraph(eg, jobVertices);
+ }
+
+ /**
+ * Verifies that building an {@link ExecutionGraph} from a {@link JobGraph} with
+ * parallelism higher than the maximum parallelism fails.
+ *
+ * TODO this test is ignored, since currently the rescale does not properly fail when rescaling to DOP above max.
+ */
+ @Ignore
+ @Test
+ public void testExecutionGraphConstructionFailsRescaleDopExceedMaxParallelism() throws Exception {
+
+ final Configuration config = new Configuration();
+
+ final JobVertex[] jobVertices = createVerticesForSimpleBipartiteJobGraph();
+ final JobGraph jobGraph = new JobGraph(jobVertices);
+
+ for (JobVertex jv : jobVertices) {
+ jv.setParallelism(ExecutionConfig.PARALLELISM_AUTO_MAX);
+ jv.setMaxParallelism(5);
+ }
+
+ try {
+ ExecutionGraphBuilder.buildGraph(
+ null,
+ jobGraph,
+ config,
+ TestingUtils.defaultExecutor(),
+ TestingUtils.defaultExecutor(),
+ new Scheduler(TestingUtils.defaultExecutionContext()),
+ Thread.currentThread().getContextClassLoader(),
+ new StandaloneCheckpointRecoveryFactory(),
+ AkkaUtils.getDefaultTimeout(),
+ new NoRestartStrategy(),
+ new UnregisteredMetricsGroup(),
+ 10, // this should fail because 10 is larger than the max parallelism 5
+ TEST_LOGGER);
+
+ fail("Building the ExecutionGraph with a parallelism higher than the max parallelism should fail.");
+ } catch (Exception e) {
+ // expected, ignore
+ }
+ }
+
+ private static JobVertex[] createVerticesForSimpleBipartiteJobGraph() {
+ JobVertex v1 = new JobVertex("vertex1");
+ JobVertex v2 = new JobVertex("vertex2");
+ JobVertex v3 = new JobVertex("vertex3");
+ JobVertex v4 = new JobVertex("vertex4");
+ JobVertex v5 = new JobVertex("vertex5");
+
+ JobVertex[] jobVertices = new JobVertex[]{v1, v2, v3, v4, v5};
+
+ for (JobVertex jobVertex : jobVertices) {
+ jobVertex.setInvokableClass(AbstractInvokable.class);
+ }
+
+ v2.connectNewDataSetAsInput(v1, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
+ v4.connectNewDataSetAsInput(v2, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
+ v4.connectNewDataSetAsInput(v3, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
+ v5.connectNewDataSetAsInput(v4, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
+ v5.connectNewDataSetAsInput(v3, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
+
+ return jobVertices;
+ }
+
+ private static void verifyGeneratedExecutionGraphOfSimpleBitartiteJobGraph(
+ ExecutionGraph generatedExecutionGraph, JobVertex[] jobVertices) {
+
+ ExecutionGraphTestUtils.verifyGeneratedExecutionJobVertex(
+ generatedExecutionGraph, jobVertices[0],
+ null, Collections.singletonList(jobVertices[1]));
+
+ ExecutionGraphTestUtils.verifyGeneratedExecutionJobVertex(
+ generatedExecutionGraph, jobVertices[1],
+ Collections.singletonList(jobVertices[0]), Collections.singletonList(jobVertices[3]));
+
+ ExecutionGraphTestUtils.verifyGeneratedExecutionJobVertex(
+ generatedExecutionGraph, jobVertices[2],
+ null, Arrays.asList(jobVertices[3], jobVertices[4]));
+
+ ExecutionGraphTestUtils.verifyGeneratedExecutionJobVertex(
+ generatedExecutionGraph, jobVertices[3],
+ Arrays.asList(jobVertices[1], jobVertices[2]), Collections.singletonList(jobVertices[4]));
+
+ ExecutionGraphTestUtils.verifyGeneratedExecutionJobVertex(
+ generatedExecutionGraph, jobVertices[4],
+ Arrays.asList(jobVertices[3], jobVertices[2]), null);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/5fe31cbf/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java
index b1ee3cc..044cb52 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java
@@ -64,8 +64,10 @@ import akka.actor.Status;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
import java.lang.reflect.Field;
import java.net.InetAddress;
+import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeoutException;
@@ -74,6 +76,8 @@ import scala.concurrent.ExecutionContext$;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
@@ -476,4 +480,77 @@ public class ExecutionGraphTestUtils {
public static ExecutionJobVertex getExecutionVertex(JobVertexID id) throws Exception {
return getExecutionVertex(id, TestingUtils.defaultExecutor());
}
+
+ // ------------------------------------------------------------------------
+ // graph vertex verifications
+ // ------------------------------------------------------------------------
+
+ /**
+ * Verifies the generated {@link ExecutionJobVertex} for a given {@link JobVertex} in a {@link ExecutionGraph}
+ *
+ * @param executionGraph the generated execution graph
+ * @param originJobVertex the vertex to verify for
+ * @param inputJobVertices upstream vertices of the verified vertex, used to check inputs of generated vertex
+ * @param outputJobVertices downstream vertices of the verified vertex, used to
+ * check produced data sets of generated vertex
+ */
+ public static void verifyGeneratedExecutionJobVertex(
+ ExecutionGraph executionGraph,
+ JobVertex originJobVertex,
+ @Nullable List<JobVertex> inputJobVertices,
+ @Nullable List<JobVertex> outputJobVertices) {
+
+ ExecutionJobVertex ejv = executionGraph.getAllVertices().get(originJobVertex.getID());
+ assertNotNull(ejv);
+
+ // verify basic properties
+ assertEquals(originJobVertex.getParallelism(), ejv.getParallelism());
+ assertEquals(executionGraph.getJobID(), ejv.getJobId());
+ assertEquals(originJobVertex.getID(), ejv.getJobVertexId());
+ assertEquals(originJobVertex, ejv.getJobVertex());
+
+ // verify produced data sets
+ if (outputJobVertices == null) {
+ assertEquals(0, ejv.getProducedDataSets().length);
+ } else {
+ assertEquals(outputJobVertices.size(), ejv.getProducedDataSets().length);
+ for (int i = 0; i < outputJobVertices.size(); i++) {
+ assertEquals(originJobVertex.getProducedDataSets().get(i).getId(), ejv.getProducedDataSets()[i].getId());
+ assertEquals(originJobVertex.getParallelism(), ejv.getProducedDataSets()[0].getPartitions().length);
+ }
+ }
+
+ // verify task vertices for their basic properties and their inputs
+ assertEquals(originJobVertex.getParallelism(), ejv.getTaskVertices().length);
+
+ int subtaskIndex = 0;
+ for (ExecutionVertex ev : ejv.getTaskVertices()) {
+ assertEquals(executionGraph.getJobID(), ev.getJobId());
+ assertEquals(originJobVertex.getID(), ev.getJobvertexId());
+
+ assertEquals(originJobVertex.getParallelism(), ev.getTotalNumberOfParallelSubtasks());
+ assertEquals(subtaskIndex, ev.getParallelSubtaskIndex());
+
+ if (inputJobVertices == null) {
+ assertEquals(0, ev.getNumberOfInputs());
+ } else {
+ assertEquals(inputJobVertices.size(), ev.getNumberOfInputs());
+
+ for (int i = 0; i < inputJobVertices.size(); i++) {
+ ExecutionEdge[] inputEdges = ev.getInputEdges(i);
+ assertEquals(inputJobVertices.get(i).getParallelism(), inputEdges.length);
+
+ int expectedPartitionNum = 0;
+ for (ExecutionEdge inEdge : inputEdges) {
+ assertEquals(i, inEdge.getInputNum());
+ assertEquals(expectedPartitionNum, inEdge.getSource().getPartitionNumber());
+
+ expectedPartitionNum++;
+ }
+ }
+ }
+
+ subtaskIndex++;
+ }
+ }
}
[2/3] flink git commit: [FLINK-7124] [flip6] Introduce parallelism <=
max parallelism check into ExecutionJobVertex
Posted by tr...@apache.org.
[FLINK-7124] [flip6] Introduce parallelism <= max parallelism check into ExecutionJobVertex
Check that the parallelism is smaller than the max parallelism when creating an
ExecutionJobVertex.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/aee578b1
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/aee578b1
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/aee578b1
Branch: refs/heads/master
Commit: aee578b14dab17d56e7ee6e08f02c9d858ebec5b
Parents: 5fe31cb
Author: Till Rohrmann <tr...@apache.org>
Authored: Fri Feb 2 09:57:14 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Sat Feb 10 23:43:52 2018 +0100
----------------------------------------------------------------------
.../executiongraph/ExecutionGraphBuilder.java | 2 +-
.../executiongraph/ExecutionJobVertex.java | 33 +++--
.../ExecutionGraphRescalingTest.java | 122 +++++++++----------
.../executiongraph/ExecutionJobVertexTest.java | 18 +--
4 files changed, 94 insertions(+), 81 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/aee578b1/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraphBuilder.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraphBuilder.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraphBuilder.java
index fe70bb6..cd719be 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraphBuilder.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraphBuilder.java
@@ -43,13 +43,13 @@ import org.apache.flink.runtime.executiongraph.metrics.NumberOfFullRestartsGauge
import org.apache.flink.runtime.executiongraph.metrics.RestartTimeGauge;
import org.apache.flink.runtime.executiongraph.metrics.UpTimeGauge;
import org.apache.flink.runtime.executiongraph.restart.RestartStrategy;
-import org.apache.flink.runtime.jobmaster.slotpool.SlotProvider;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.jsonplan.JsonPlanGenerator;
import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
import org.apache.flink.runtime.jobgraph.tasks.JobCheckpointingSettings;
+import org.apache.flink.runtime.jobmaster.slotpool.SlotProvider;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.StateBackendLoader;
import org.apache.flink.util.DynamicCodeLoadingException;
http://git-wip-us.apache.org/repos/asf/flink/blob/aee578b1/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertex.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertex.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertex.java
index bb5ad28..e07c71e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertex.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertex.java
@@ -35,7 +35,6 @@ import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
import org.apache.flink.runtime.blob.BlobWriter;
import org.apache.flink.runtime.blob.PermanentBlobKey;
import org.apache.flink.runtime.execution.ExecutionState;
-import org.apache.flink.runtime.jobmaster.slotpool.SlotProvider;
import org.apache.flink.runtime.jobgraph.IntermediateDataSet;
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
import org.apache.flink.runtime.jobgraph.JobEdge;
@@ -45,6 +44,7 @@ import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.jobmanager.scheduler.CoLocationGroup;
import org.apache.flink.runtime.jobmanager.scheduler.LocationPreferenceConstraint;
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
+import org.apache.flink.runtime.jobmaster.slotpool.SlotProvider;
import org.apache.flink.runtime.state.KeyGroupRangeAssignment;
import org.apache.flink.types.Either;
import org.apache.flink.util.Preconditions;
@@ -73,7 +73,7 @@ import java.util.concurrent.CompletableFuture;
*/
public class ExecutionJobVertex implements AccessExecutionJobVertex, Archiveable<ArchivedExecutionJobVertex> {
- /** Use the same log for all ExecutionGraph classes */
+ /** Use the same log for all ExecutionGraph classes. */
private static final Logger LOG = ExecutionGraph.LOG;
public static final int VALUE_NOT_SET = -1;
@@ -87,7 +87,7 @@ public class ExecutionJobVertex implements AccessExecutionJobVertex, Archiveable
/**
* The IDs of all operators contained in this execution job vertex.
*
- * The ID's are stored depth-first post-order; for the forking chain below the ID's would be stored as [D, E, B, C, A].
+ * <p>The ID's are stored depth-first post-order; for the forking chain below the ID's would be stored as [D, E, B, C, A].
* A - B - D
* \ \
* C E
@@ -98,7 +98,7 @@ public class ExecutionJobVertex implements AccessExecutionJobVertex, Archiveable
/**
* The alternative IDs of all operators contained in this execution job vertex.
*
- * The ID's are in the same order as {@link ExecutionJobVertex#operatorIDs}.
+ * <p>The ID's are in the same order as {@link ExecutionJobVertex#operatorIDs}.
*/
private final List<OperatorID> userDefinedOperatorIds;
@@ -162,22 +162,31 @@ public class ExecutionJobVertex implements AccessExecutionJobVertex, Archiveable
if (graph == null || jobVertex == null) {
throw new NullPointerException();
}
-
+
this.graph = graph;
this.jobVertex = jobVertex;
int vertexParallelism = jobVertex.getParallelism();
int numTaskVertices = vertexParallelism > 0 ? vertexParallelism : defaultParallelism;
- this.parallelism = numTaskVertices;
-
final int configuredMaxParallelism = jobVertex.getMaxParallelism();
this.maxParallelismConfigured = (VALUE_NOT_SET != configuredMaxParallelism);
// if no max parallelism was configured by the user, we calculate and set a default
setMaxParallelismInternal(maxParallelismConfigured ?
- configuredMaxParallelism : KeyGroupRangeAssignment.computeDefaultMaxParallelism(parallelism));
+ configuredMaxParallelism : KeyGroupRangeAssignment.computeDefaultMaxParallelism(numTaskVertices));
+
+ // verify that our parallelism is not higher than the maximum parallelism
+ if (numTaskVertices > maxParallelism) {
+ throw new JobException(
+ String.format("Vertex %s's parallelism (%s) is higher than the max parallelism (%s). Please lower the parallelism or increase the max parallelism.",
+ jobVertex.getName(),
+ numTaskVertices,
+ maxParallelism));
+ }
+
+ this.parallelism = numTaskVertices;
this.serializedTaskInformation = null;
@@ -186,16 +195,16 @@ public class ExecutionJobVertex implements AccessExecutionJobVertex, Archiveable
this.userDefinedOperatorIds = Collections.unmodifiableList(jobVertex.getUserDefinedOperatorIDs());
this.inputs = new ArrayList<>(jobVertex.getInputs().size());
-
+
// take the sharing group
this.slotSharingGroup = jobVertex.getSlotSharingGroup();
this.coLocationGroup = jobVertex.getCoLocationGroup();
-
+
// setup the coLocation group
if (coLocationGroup != null && slotSharingGroup == null) {
throw new JobException("Vertex uses a co-location constraint without using slot sharing");
}
-
+
// create the intermediate results
this.producedDataSets = new IntermediateResult[jobVertex.getNumberOfProducedIntermediateDataSets()];
@@ -239,7 +248,7 @@ public class ExecutionJobVertex implements AccessExecutionJobVertex, Archiveable
try {
@SuppressWarnings("unchecked")
InputSplitSource<InputSplit> splitSource = (InputSplitSource<InputSplit>) jobVertex.getInputSplitSource();
-
+
if (splitSource != null) {
Thread currentThread = Thread.currentThread();
ClassLoader oldContextClassLoader = currentThread.getContextClassLoader();
http://git-wip-us.apache.org/repos/asf/flink/blob/aee578b1/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRescalingTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRescalingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRescalingTest.java
index e6cc908..f4a9f25 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRescalingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRescalingTest.java
@@ -18,10 +18,11 @@
package org.apache.flink.runtime.executiongraph;
-import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.runtime.JobException;
import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.blob.VoidBlobWriter;
import org.apache.flink.runtime.checkpoint.StandaloneCheckpointRecoveryFactory;
import org.apache.flink.runtime.executiongraph.restart.NoRestartStrategy;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
@@ -31,8 +32,8 @@ import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.jobmanager.scheduler.Scheduler;
import org.apache.flink.runtime.testingUtils.TestingUtils;
+import org.apache.flink.util.TestLogger;
-import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,14 +41,15 @@ import org.slf4j.LoggerFactory;
import java.util.Arrays;
import java.util.Collections;
-import static org.junit.Assert.assertEquals;
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
/**
* This class contains tests that verify when rescaling a {@link JobGraph},
* constructed {@link ExecutionGraph}s are correct.
*/
-public class ExecutionGraphRescalingTest {
+public class ExecutionGraphRescalingTest extends TestLogger {
private static final Logger TEST_LOGGER = LoggerFactory.getLogger(ExecutionGraphRescalingTest.class);
@@ -56,72 +58,67 @@ public class ExecutionGraphRescalingTest {
final Configuration config = new Configuration();
- final JobVertex[] jobVertices = createVerticesForSimpleBipartiteJobGraph();
+ final int initialParallelism = 5;
+ final int maxParallelism = 10;
+ final JobVertex[] jobVertices = createVerticesForSimpleBipartiteJobGraph(initialParallelism, maxParallelism);
final JobGraph jobGraph = new JobGraph(jobVertices);
- // TODO rescaling the JobGraph is currently only supported if the
- // TODO configured parallelism is ExecutionConfig.PARALLELISM_AUTO_MAX.
- // TODO this limitation should be removed.
- for (JobVertex jv : jobVertices) {
- jv.setParallelism(ExecutionConfig.PARALLELISM_AUTO_MAX);
- }
-
ExecutionGraph eg = ExecutionGraphBuilder.buildGraph(
- null,
- jobGraph,
- config,
- TestingUtils.defaultExecutor(),
- TestingUtils.defaultExecutor(),
- new Scheduler(TestingUtils.defaultExecutionContext()),
- Thread.currentThread().getContextClassLoader(),
- new StandaloneCheckpointRecoveryFactory(),
- AkkaUtils.getDefaultTimeout(),
- new NoRestartStrategy(),
- new UnregisteredMetricsGroup(),
- 5,
- TEST_LOGGER);
+ null,
+ jobGraph,
+ config,
+ TestingUtils.defaultExecutor(),
+ TestingUtils.defaultExecutor(),
+ new Scheduler(TestingUtils.defaultExecutionContext()),
+ Thread.currentThread().getContextClassLoader(),
+ new StandaloneCheckpointRecoveryFactory(),
+ AkkaUtils.getDefaultTimeout(),
+ new NoRestartStrategy(),
+ new UnregisteredMetricsGroup(),
+ -1,
+ VoidBlobWriter.getInstance(),
+ TEST_LOGGER);
for (JobVertex jv : jobVertices) {
- assertEquals(5, jv.getParallelism());
+ assertThat(jv.getParallelism(), is(initialParallelism));
}
verifyGeneratedExecutionGraphOfSimpleBitartiteJobGraph(eg, jobVertices);
- // --- verify scaling up works correctly ---
+ // --- verify scaling down works correctly ---
+
+ final int scaleDownParallelism = 1;
- // TODO rescaling the JobGraph is currently only supported if the
- // TODO configured parallelism is ExecutionConfig.PARALLELISM_AUTO_MAX.
- // TODO this limitation should be removed.
for (JobVertex jv : jobVertices) {
- jv.setParallelism(ExecutionConfig.PARALLELISM_AUTO_MAX);
+ jv.setParallelism(scaleDownParallelism);
}
eg = ExecutionGraphBuilder.buildGraph(
- null,
- jobGraph,
- config,
- TestingUtils.defaultExecutor(),
- TestingUtils.defaultExecutor(),
- new Scheduler(TestingUtils.defaultExecutionContext()),
- Thread.currentThread().getContextClassLoader(),
- new StandaloneCheckpointRecoveryFactory(),
- AkkaUtils.getDefaultTimeout(),
- new NoRestartStrategy(),
- new UnregisteredMetricsGroup(),
- 10,
- TEST_LOGGER);
+ null,
+ jobGraph,
+ config,
+ TestingUtils.defaultExecutor(),
+ TestingUtils.defaultExecutor(),
+ new Scheduler(TestingUtils.defaultExecutionContext()),
+ Thread.currentThread().getContextClassLoader(),
+ new StandaloneCheckpointRecoveryFactory(),
+ AkkaUtils.getDefaultTimeout(),
+ new NoRestartStrategy(),
+ new UnregisteredMetricsGroup(),
+ -1,
+ VoidBlobWriter.getInstance(),
+ TEST_LOGGER);
for (JobVertex jv : jobVertices) {
- assertEquals(10, jv.getParallelism());
+ assertThat(jv.getParallelism(), is(1));
}
verifyGeneratedExecutionGraphOfSimpleBitartiteJobGraph(eg, jobVertices);
- // --- verify down scaling works correctly ---
+ // --- verify scaling up works correctly ---
+
+ final int scaleUpParallelism = 10;
- // TODO rescaling the JobGraph is currently only supported if the
- // TODO configured parallelism is ExecutionConfig.PARALLELISM_AUTO_MAX.
- // TODO this limitation should be removed.
for (JobVertex jv : jobVertices) {
- jv.setParallelism(ExecutionConfig.PARALLELISM_AUTO_MAX);
+ jv.setParallelism(scaleUpParallelism);
}
eg = ExecutionGraphBuilder.buildGraph(
@@ -136,11 +133,12 @@ public class ExecutionGraphRescalingTest {
AkkaUtils.getDefaultTimeout(),
new NoRestartStrategy(),
new UnregisteredMetricsGroup(),
- 2,
+ -1,
+ VoidBlobWriter.getInstance(),
TEST_LOGGER);
for (JobVertex jv : jobVertices) {
- assertEquals(2, jv.getParallelism());
+ assertThat(jv.getParallelism(), is(scaleUpParallelism));
}
verifyGeneratedExecutionGraphOfSimpleBitartiteJobGraph(eg, jobVertices);
}
@@ -148,24 +146,23 @@ public class ExecutionGraphRescalingTest {
/**
* Verifies that building an {@link ExecutionGraph} from a {@link JobGraph} with
* parallelism higher than the maximum parallelism fails.
- *
- * TODO this test is ignored, since currently the rescale does not properly fail when rescaling to DOP above max.
*/
- @Ignore
@Test
public void testExecutionGraphConstructionFailsRescaleDopExceedMaxParallelism() throws Exception {
final Configuration config = new Configuration();
- final JobVertex[] jobVertices = createVerticesForSimpleBipartiteJobGraph();
+ final int initialParallelism = 1;
+ final int maxParallelism = 10;
+ final JobVertex[] jobVertices = createVerticesForSimpleBipartiteJobGraph(initialParallelism, maxParallelism);
final JobGraph jobGraph = new JobGraph(jobVertices);
for (JobVertex jv : jobVertices) {
- jv.setParallelism(ExecutionConfig.PARALLELISM_AUTO_MAX);
- jv.setMaxParallelism(5);
+ jv.setParallelism(maxParallelism + 1);
}
try {
+ // this should fail since we set the parallelism to maxParallelism + 1
ExecutionGraphBuilder.buildGraph(
null,
jobGraph,
@@ -178,16 +175,17 @@ public class ExecutionGraphRescalingTest {
AkkaUtils.getDefaultTimeout(),
new NoRestartStrategy(),
new UnregisteredMetricsGroup(),
- 10, // this should fail because 10 is larger than the max parallelism 5
+ -1,
+ VoidBlobWriter.getInstance(),
TEST_LOGGER);
fail("Building the ExecutionGraph with a parallelism higher than the max parallelism should fail.");
- } catch (Exception e) {
+ } catch (JobException e) {
// expected, ignore
}
}
- private static JobVertex[] createVerticesForSimpleBipartiteJobGraph() {
+ private static JobVertex[] createVerticesForSimpleBipartiteJobGraph(int parallelism, int maxParallelism) {
JobVertex v1 = new JobVertex("vertex1");
JobVertex v2 = new JobVertex("vertex2");
JobVertex v3 = new JobVertex("vertex3");
@@ -198,6 +196,8 @@ public class ExecutionGraphRescalingTest {
for (JobVertex jobVertex : jobVertices) {
jobVertex.setInvokableClass(AbstractInvokable.class);
+ jobVertex.setParallelism(parallelism);
+ jobVertex.setMaxParallelism(maxParallelism);
}
v2.connectNewDataSetAsInput(v1, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED);
http://git-wip-us.apache.org/repos/asf/flink/blob/aee578b1/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertexTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertexTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertexTest.java
index f0f6248..78c87b7 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertexTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertexTest.java
@@ -23,11 +23,10 @@ import org.apache.flink.runtime.JobException;
import org.apache.flink.runtime.concurrent.Executors;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
+
import org.junit.Assert;
import org.junit.Test;
-import java.io.IOException;
-
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -62,9 +61,14 @@ public class ExecutionJobVertexTest {
} catch (IllegalArgumentException ignore) {
}
- // test configured / trumps computed default
- executionJobVertex = createExecutionJobVertex(172, 4);
- Assert.assertEquals(4, executionJobVertex.getMaxParallelism());
+ // parallelism must be smaller than the max parallelism
+ try {
+ createExecutionJobVertex(172, 4);
+ Assert.fail("We should not be able to create an ExecutionJobVertex which " +
+ "has a smaller max parallelism than parallelism.");
+ } catch (JobException ignored) {
+ // expected
+ }
// test configured / trumps computed default
@@ -120,7 +124,7 @@ public class ExecutionJobVertexTest {
private static ExecutionJobVertex createExecutionJobVertex(
int parallelism,
- int preconfiguredMaxParallelism) throws JobException, IOException {
+ int preconfiguredMaxParallelism) throws JobException {
JobVertex jobVertex = new JobVertex("testVertex");
jobVertex.setInvokableClass(AbstractInvokable.class);
@@ -137,4 +141,4 @@ public class ExecutionJobVertexTest {
return executionJobVertex;
}
-}
\ No newline at end of file
+}
[3/3] flink git commit: [hotfix] Refactor graph verification code in
ExecutionGraphConstructionTest
Posted by tr...@apache.org.
[hotfix] Refactor graph verification code in ExecutionGraphConstructionTest
The refactoring resuses utility methods in ExecutionGraphTestUtils to
verify constructed ExecutionGraphs.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/556ea8a7
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/556ea8a7
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/556ea8a7
Branch: refs/heads/master
Commit: 556ea8a71731218096f013c5b06af57fa6404560
Parents: aee578b
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Authored: Thu Aug 10 13:57:09 2017 +0800
Committer: Till Rohrmann <tr...@apache.org>
Committed: Sat Feb 10 23:43:53 2018 +0100
----------------------------------------------------------------------
.../ExecutionGraphConstructionTest.java | 212 +------------------
1 file changed, 10 insertions(+), 202 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/556ea8a7/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphConstructionTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphConstructionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphConstructionTest.java
index e1bad56..00d1ac0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphConstructionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphConstructionTest.java
@@ -40,9 +40,9 @@ import org.mockito.Matchers;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
-import java.util.Map;
import java.util.Set;
import org.apache.flink.configuration.Configuration;
@@ -50,14 +50,12 @@ import org.apache.flink.core.io.InputSplit;
import org.apache.flink.core.io.InputSplitAssigner;
import org.apache.flink.core.io.InputSplitSource;
import org.apache.flink.runtime.JobException;
-import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.IntermediateDataSet;
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobmanager.scheduler.CoLocationConstraint;
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
@@ -130,7 +128,7 @@ public class ExecutionGraphConstructionTest {
fail("Job failed with exception: " + e.getMessage());
}
- verifyTestGraph(eg, jobId, v1, v2, v3, v4, v5);
+ verifyTestGraph(eg, v1, v2, v3, v4, v5);
}
@Test
@@ -207,7 +205,7 @@ public class ExecutionGraphConstructionTest {
}
// verify
- verifyTestGraph(eg, jobId, v1, v2, v3, v4, v5);
+ verifyTestGraph(eg, v1, v2, v3, v4, v5);
}
@Test
@@ -283,208 +281,18 @@ public class ExecutionGraphConstructionTest {
}
// verify
- verifyTestGraph(eg, jobId, v1, v2, v3, v4, v5);
+ verifyTestGraph(eg, v1, v2, v3, v4, v5);
}
- private void verifyTestGraph(ExecutionGraph eg, JobID jobId,
+ private void verifyTestGraph(ExecutionGraph eg,
JobVertex v1, JobVertex v2, JobVertex v3,
JobVertex v4, JobVertex v5)
{
- Map<JobVertexID, ExecutionJobVertex> vertices = eg.getAllVertices();
-
- // verify v1
- {
- ExecutionJobVertex e1 = vertices.get(v1.getID());
- assertNotNull(e1);
-
- // basic properties
- assertEquals(v1.getParallelism(), e1.getParallelism());
- assertEquals(v1.getID(), e1.getJobVertexId());
- assertEquals(jobId, e1.getJobId());
- assertEquals(v1, e1.getJobVertex());
-
- // produced data sets
- assertEquals(1, e1.getProducedDataSets().length);
- assertEquals(v1.getProducedDataSets().get(0).getId(), e1.getProducedDataSets()[0].getId());
- assertEquals(v1.getParallelism(), e1.getProducedDataSets()[0].getPartitions().length);
-
- // task vertices
- assertEquals(v1.getParallelism(), e1.getTaskVertices().length);
-
- int num = 0;
- for (ExecutionVertex ev : e1.getTaskVertices()) {
- assertEquals(jobId, ev.getJobId());
- assertEquals(v1.getID(), ev.getJobvertexId());
-
- assertEquals(v1.getParallelism(), ev.getTotalNumberOfParallelSubtasks());
- assertEquals(num++, ev.getParallelSubtaskIndex());
-
- assertEquals(0, ev.getNumberOfInputs());
-
- assertTrue(ev.getStateTimestamp(ExecutionState.CREATED) > 0);
- }
- }
-
- // verify v2
- {
- ExecutionJobVertex e2 = vertices.get(v2.getID());
- assertNotNull(e2);
-
- // produced data sets
- assertEquals(1, e2.getProducedDataSets().length);
- assertEquals(v2.getProducedDataSets().get(0).getId(), e2.getProducedDataSets()[0].getId());
- assertEquals(v2.getParallelism(), e2.getProducedDataSets()[0].getPartitions().length);
-
- // task vertices
- assertEquals(v2.getParallelism(), e2.getTaskVertices().length);
-
- int num = 0;
- for (ExecutionVertex ev : e2.getTaskVertices()) {
- assertEquals(jobId, ev.getJobId());
- assertEquals(v2.getID(), ev.getJobvertexId());
-
- assertEquals(v2.getParallelism(), ev.getTotalNumberOfParallelSubtasks());
- assertEquals(num++, ev.getParallelSubtaskIndex());
-
- assertEquals(1, ev.getNumberOfInputs());
- ExecutionEdge[] inputs = ev.getInputEdges(0);
- assertEquals(v1.getParallelism(), inputs.length);
-
- int sumOfPartitions = 0;
- for (ExecutionEdge inEdge : inputs) {
- assertEquals(0,inEdge.getInputNum());
- sumOfPartitions += inEdge.getSource().getPartitionNumber();
- }
-
- assertEquals(10, sumOfPartitions);
- }
- }
-
- // verify v3
- {
- ExecutionJobVertex e3 = vertices.get(v3.getID());
- assertNotNull(e3);
-
- // produced data sets
- assertEquals(2, e3.getProducedDataSets().length);
- assertEquals(v3.getProducedDataSets().get(0).getId(), e3.getProducedDataSets()[0].getId());
- assertEquals(v3.getProducedDataSets().get(1).getId(), e3.getProducedDataSets()[1].getId());
- assertEquals(v3.getParallelism(), e3.getProducedDataSets()[0].getPartitions().length);
- assertEquals(v3.getParallelism(), e3.getProducedDataSets()[1].getPartitions().length);
-
- // task vertices
- assertEquals(v3.getParallelism(), e3.getTaskVertices().length);
-
- int num = 0;
- for (ExecutionVertex ev : e3.getTaskVertices()) {
- assertEquals(jobId, ev.getJobId());
- assertEquals(v3.getID(), ev.getJobvertexId());
-
- assertEquals(v3.getParallelism(), ev.getTotalNumberOfParallelSubtasks());
- assertEquals(num++, ev.getParallelSubtaskIndex());
-
- assertEquals(0, ev.getNumberOfInputs());
- }
- }
-
- // verify v4
- {
- ExecutionJobVertex e4 = vertices.get(v4.getID());
- assertNotNull(e4);
-
- // produced data sets
- assertEquals(1, e4.getProducedDataSets().length);
- assertEquals(v4.getProducedDataSets().get(0).getId(), e4.getProducedDataSets()[0].getId());
-
- // task vertices
- assertEquals(v4.getParallelism(), e4.getTaskVertices().length);
-
- int num = 0;
- for (ExecutionVertex ev : e4.getTaskVertices()) {
- assertEquals(jobId, ev.getJobId());
- assertEquals(v4.getID(), ev.getJobvertexId());
-
- assertEquals(v4.getParallelism(), ev.getTotalNumberOfParallelSubtasks());
- assertEquals(num++, ev.getParallelSubtaskIndex());
-
- assertEquals(2, ev.getNumberOfInputs());
- // first input
- {
- ExecutionEdge[] inputs = ev.getInputEdges(0);
- assertEquals(v2.getParallelism(), inputs.length);
-
- int sumOfPartitions = 0;
- for (ExecutionEdge inEdge : inputs) {
- assertEquals(0, inEdge.getInputNum());
- sumOfPartitions += inEdge.getSource().getPartitionNumber();
- }
-
- assertEquals(21, sumOfPartitions);
- }
- // second input
- {
- ExecutionEdge[] inputs = ev.getInputEdges(1);
- assertEquals(v3.getParallelism(), inputs.length);
-
- int sumOfPartitions = 0;
- for (ExecutionEdge inEdge : inputs) {
- assertEquals(1, inEdge.getInputNum());
- sumOfPartitions += inEdge.getSource().getPartitionNumber();
- }
-
- assertEquals(1, sumOfPartitions);
- }
- }
- }
-
- // verify v5
- {
- ExecutionJobVertex e5 = vertices.get(v5.getID());
- assertNotNull(e5);
-
- // produced data sets
- assertEquals(0, e5.getProducedDataSets().length);
-
- // task vertices
- assertEquals(v5.getParallelism(), e5.getTaskVertices().length);
-
- int num = 0;
- for (ExecutionVertex ev : e5.getTaskVertices()) {
- assertEquals(jobId, ev.getJobId());
- assertEquals(v5.getID(), ev.getJobvertexId());
-
- assertEquals(v5.getParallelism(), ev.getTotalNumberOfParallelSubtasks());
- assertEquals(num++, ev.getParallelSubtaskIndex());
-
- assertEquals(2, ev.getNumberOfInputs());
- // first input
- {
- ExecutionEdge[] inputs = ev.getInputEdges(0);
- assertEquals(v4.getParallelism(), inputs.length);
-
- int sumOfPartitions = 0;
- for (ExecutionEdge inEdge : inputs) {
- assertEquals(0, inEdge.getInputNum());
- sumOfPartitions += inEdge.getSource().getPartitionNumber();
- }
-
- assertEquals(55, sumOfPartitions);
- }
- // second input
- {
- ExecutionEdge[] inputs = ev.getInputEdges(1);
- assertEquals(v3.getParallelism(), inputs.length);
-
- int sumOfPartitions = 0;
- for (ExecutionEdge inEdge : inputs) {
- assertEquals(1, inEdge.getInputNum());
- sumOfPartitions += inEdge.getSource().getPartitionNumber();
- }
-
- assertEquals(1, sumOfPartitions);
- }
- }
- }
+ ExecutionGraphTestUtils.verifyGeneratedExecutionJobVertex(eg, v1, null, Collections.singletonList(v2));
+ ExecutionGraphTestUtils.verifyGeneratedExecutionJobVertex(eg, v2, Collections.singletonList(v1), Collections.singletonList(v4));
+ ExecutionGraphTestUtils.verifyGeneratedExecutionJobVertex(eg, v3, null, Arrays.asList(v4, v5));
+ ExecutionGraphTestUtils.verifyGeneratedExecutionJobVertex(eg, v4, Arrays.asList(v2, v3), Collections.singletonList(v5));
+ ExecutionGraphTestUtils.verifyGeneratedExecutionJobVertex(eg, v5, Arrays.asList(v4, v3), null);
}
@Test