You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by mx...@apache.org on 2015/06/16 17:20:08 UTC
[1/3] flink git commit: [FLINK-2120][runtime] rename
AbstractJobVertex to JobVertex
Repository: flink
Updated Branches:
refs/heads/master 20d7e5ad3 -> 1bd0af738
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/IterateTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/IterateTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/IterateTest.java
index 9c6ff5c..a9c0ee3 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/IterateTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/IterateTest.java
@@ -25,7 +25,7 @@ import java.util.Collections;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.IterativeDataStream;
@@ -123,12 +123,12 @@ public class IterateTest {
JobGraph graph = env.getStreamGraph().getJobGraph();
- AbstractJobVertex itSource = null;
- AbstractJobVertex itSink = null;
- AbstractJobVertex headOp = null;
- AbstractJobVertex tailOp = null;
+ JobVertex itSource = null;
+ JobVertex itSink = null;
+ JobVertex headOp = null;
+ JobVertex tailOp = null;
- for (AbstractJobVertex vertex : graph.getVertices()) {
+ for (JobVertex vertex : graph.getVertices()) {
if (vertex.getName().contains("IterationSource")) {
itSource = vertex;
} else if (vertex.getName().contains("IterationSink")) {
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/graph/SlotAllocationTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/graph/SlotAllocationTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/graph/SlotAllocationTest.java
index 94636f7..767eaa4 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/graph/SlotAllocationTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/graph/SlotAllocationTest.java
@@ -22,7 +22,7 @@ import static org.junit.Assert.*;
import java.util.List;
import org.apache.flink.api.common.functions.FilterFunction;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.junit.Test;
@@ -49,7 +49,7 @@ public class SlotAllocationTest {
JobGraph jobGraph = env.getStreamGraph().getJobGraph();
- List<AbstractJobVertex> vertices = jobGraph.getVerticesSortedTopologicallyFromSources();
+ List<JobVertex> vertices = jobGraph.getVerticesSortedTopologicallyFromSources();
assertEquals(vertices.get(0).getSlotSharingGroup(), vertices.get(2).getSlotSharingGroup());
assertNotEquals(vertices.get(0).getSlotSharingGroup(), vertices.get(1)
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-tests/src/test/java/org/apache/flink/test/broadcastvars/BroadcastVarsNepheleITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/broadcastvars/BroadcastVarsNepheleITCase.java b/flink-tests/src/test/java/org/apache/flink/test/broadcastvars/BroadcastVarsNepheleITCase.java
index 34d4133..9366058 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/broadcastvars/BroadcastVarsNepheleITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/broadcastvars/BroadcastVarsNepheleITCase.java
@@ -35,7 +35,7 @@ import org.apache.flink.api.java.record.io.CsvInputFormat;
import org.apache.flink.api.java.record.io.CsvOutputFormat;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.InputFormatVertex;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -262,8 +262,8 @@ public class BroadcastVarsNepheleITCase extends RecordAPITestBase {
return modelsInput;
}
- private static AbstractJobVertex createMapper(JobGraph jobGraph, int numSubTasks, TypeSerializerFactory<?> serializer) {
- AbstractJobVertex pointsInput = JobGraphUtils.createTask(RegularPactTask.class, "Map[DotProducts]", jobGraph, numSubTasks);
+ private static JobVertex createMapper(JobGraph jobGraph, int numSubTasks, TypeSerializerFactory<?> serializer) {
+ JobVertex pointsInput = JobGraphUtils.createTask(RegularPactTask.class, "Map[DotProducts]", jobGraph, numSubTasks);
{
TaskConfig taskConfig = new TaskConfig(pointsInput.getConfiguration());
@@ -318,7 +318,7 @@ public class BroadcastVarsNepheleITCase extends RecordAPITestBase {
// -- vertices ---------------------------------------------------------------------------------------------
InputFormatVertex points = createPointsInput(jobGraph, pointsPath, numSubTasks, serializer);
InputFormatVertex models = createModelsInput(jobGraph, centersPath, numSubTasks, serializer);
- AbstractJobVertex mapper = createMapper(jobGraph, numSubTasks, serializer);
+ JobVertex mapper = createMapper(jobGraph, numSubTasks, serializer);
OutputFormatVertex output = createOutput(jobGraph, resultPath, numSubTasks, serializer);
// -- edges ------------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-tests/src/test/java/org/apache/flink/test/broadcastvars/KMeansIterativeNepheleITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/broadcastvars/KMeansIterativeNepheleITCase.java b/flink-tests/src/test/java/org/apache/flink/test/broadcastvars/KMeansIterativeNepheleITCase.java
index 2406d6e..61ba59a 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/broadcastvars/KMeansIterativeNepheleITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/broadcastvars/KMeansIterativeNepheleITCase.java
@@ -30,7 +30,7 @@ import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.iterative.task.IterationHeadPactTask;
import org.apache.flink.runtime.iterative.task.IterationIntermediatePactTask;
import org.apache.flink.runtime.iterative.task.IterationTailPactTask;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.InputFormatVertex;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -155,8 +155,8 @@ public class KMeansIterativeNepheleITCase extends RecordAPITestBase {
return output;
}
- private static AbstractJobVertex createIterationHead(JobGraph jobGraph, int numSubTasks, TypeSerializerFactory<?> serializer) {
- AbstractJobVertex head = JobGraphUtils.createTask(IterationHeadPactTask.class, "Iteration Head", jobGraph, numSubTasks);
+ private static JobVertex createIterationHead(JobGraph jobGraph, int numSubTasks, TypeSerializerFactory<?> serializer) {
+ JobVertex head = JobGraphUtils.createTask(IterationHeadPactTask.class, "Iteration Head", jobGraph, numSubTasks);
TaskConfig headConfig = new TaskConfig(head.getConfiguration());
headConfig.setIterationId(ITERATION_ID);
@@ -189,11 +189,11 @@ public class KMeansIterativeNepheleITCase extends RecordAPITestBase {
return head;
}
- private static AbstractJobVertex createMapper(JobGraph jobGraph, int numSubTasks, TypeSerializerFactory<?> inputSerializer,
+ private static JobVertex createMapper(JobGraph jobGraph, int numSubTasks, TypeSerializerFactory<?> inputSerializer,
TypeSerializerFactory<?> broadcastVarSerializer, TypeSerializerFactory<?> outputSerializer,
TypeComparatorFactory<?> outputComparator)
{
- AbstractJobVertex mapper = JobGraphUtils.createTask(IterationIntermediatePactTask.class,
+ JobVertex mapper = JobGraphUtils.createTask(IterationIntermediatePactTask.class,
"Map (Select nearest center)", jobGraph, numSubTasks);
TaskConfig intermediateConfig = new TaskConfig(mapper.getConfiguration());
@@ -218,12 +218,12 @@ public class KMeansIterativeNepheleITCase extends RecordAPITestBase {
return mapper;
}
- private static AbstractJobVertex createReducer(JobGraph jobGraph, int numSubTasks, TypeSerializerFactory<?> inputSerializer,
+ private static JobVertex createReducer(JobGraph jobGraph, int numSubTasks, TypeSerializerFactory<?> inputSerializer,
TypeComparatorFactory<?> inputComparator, TypeSerializerFactory<?> outputSerializer)
{
// ---------------- the tail (reduce) --------------------
- AbstractJobVertex tail = JobGraphUtils.createTask(IterationTailPactTask.class, "Reduce / Iteration Tail", jobGraph,
+ JobVertex tail = JobGraphUtils.createTask(IterationTailPactTask.class, "Reduce / Iteration Tail", jobGraph,
numSubTasks);
TaskConfig tailConfig = new TaskConfig(tail.getConfiguration());
@@ -252,8 +252,8 @@ public class KMeansIterativeNepheleITCase extends RecordAPITestBase {
return tail;
}
- private static AbstractJobVertex createSync(JobGraph jobGraph, int numIterations, int parallelism) {
- AbstractJobVertex sync = JobGraphUtils.createSync(jobGraph, parallelism);
+ private static JobVertex createSync(JobGraph jobGraph, int numIterations, int parallelism) {
+ JobVertex sync = JobGraphUtils.createSync(jobGraph, parallelism);
TaskConfig syncConfig = new TaskConfig(sync.getConfiguration());
syncConfig.setNumberOfIterations(numIterations);
syncConfig.setIterationId(ITERATION_ID);
@@ -277,12 +277,12 @@ public class KMeansIterativeNepheleITCase extends RecordAPITestBase {
InputFormatVertex points = createPointsInput(jobGraph, pointsPath, numSubTasks, serializer);
InputFormatVertex centers = createCentersInput(jobGraph, centersPath, numSubTasks, serializer);
- AbstractJobVertex head = createIterationHead(jobGraph, numSubTasks, serializer);
- AbstractJobVertex mapper = createMapper(jobGraph, numSubTasks, serializer, serializer, serializer, int0Comparator);
+ JobVertex head = createIterationHead(jobGraph, numSubTasks, serializer);
+ JobVertex mapper = createMapper(jobGraph, numSubTasks, serializer, serializer, serializer, int0Comparator);
- AbstractJobVertex reducer = createReducer(jobGraph, numSubTasks, serializer, int0Comparator, serializer);
+ JobVertex reducer = createReducer(jobGraph, numSubTasks, serializer, int0Comparator, serializer);
- AbstractJobVertex sync = createSync(jobGraph, numIterations, numSubTasks);
+ JobVertex sync = createSync(jobGraph, numIterations, numSubTasks);
OutputFormatVertex output = createOutput(jobGraph, resultPath, numSubTasks, serializer);
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-tests/src/test/java/org/apache/flink/test/failingPrograms/JobSubmissionFailsITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/failingPrograms/JobSubmissionFailsITCase.java b/flink-tests/src/test/java/org/apache/flink/test/failingPrograms/JobSubmissionFailsITCase.java
index ca1c304..2cf5598 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/failingPrograms/JobSubmissionFailsITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/failingPrograms/JobSubmissionFailsITCase.java
@@ -24,7 +24,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.client.JobExecutionException;
import org.apache.flink.runtime.client.JobSubmissionException;
import org.apache.flink.runtime.client.SerializedJobExecutionResult;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobmanager.Tasks;
import org.apache.flink.runtime.testingUtils.TestingUtils;
@@ -61,7 +61,7 @@ public class JobSubmissionFailsITCase {
cluser = new ForkableFlinkMiniCluster(config);
- final AbstractJobVertex jobVertex = new AbstractJobVertex("Working job vertex.");
+ final JobVertex jobVertex = new JobVertex("Working job vertex.");
jobVertex.setInvokableClass(Tasks.NoOpInvokable.class);
workingJobGraph = new JobGraph("Working testing job", jobVertex);
}
@@ -113,7 +113,7 @@ public class JobSubmissionFailsITCase {
@Test
public void testExceptionInInitializeOnMaster() {
try {
- final AbstractJobVertex failingJobVertex = new FailingJobVertex("Failing job vertex");
+ final JobVertex failingJobVertex = new FailingJobVertex("Failing job vertex");
failingJobVertex.setInvokableClass(Tasks.NoOpInvokable.class);
final JobGraph failingJobGraph = new JobGraph("Failing testing job", failingJobVertex);
@@ -188,7 +188,7 @@ public class JobSubmissionFailsITCase {
// --------------------------------------------------------------------------------------------
- public static class FailingJobVertex extends AbstractJobVertex {
+ public static class FailingJobVertex extends JobVertex {
private static final long serialVersionUID = -6365291240199412135L;
public FailingJobVertex(final String msg) {
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/ConnectedComponentsNepheleITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/ConnectedComponentsNepheleITCase.java b/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/ConnectedComponentsNepheleITCase.java
index 7cec122..8e9266d 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/ConnectedComponentsNepheleITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/ConnectedComponentsNepheleITCase.java
@@ -41,7 +41,7 @@ import org.apache.flink.runtime.iterative.convergence.WorksetEmptyConvergenceCri
import org.apache.flink.runtime.iterative.task.IterationHeadPactTask;
import org.apache.flink.runtime.iterative.task.IterationIntermediatePactTask;
import org.apache.flink.runtime.iterative.task.IterationTailPactTask;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.InputFormatVertex;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -221,12 +221,12 @@ public class ConnectedComponentsNepheleITCase extends RecordAPITestBase {
return edgesInput;
}
- private static AbstractJobVertex createIterationHead(JobGraph jobGraph, int numSubTasks,
+ private static JobVertex createIterationHead(JobGraph jobGraph, int numSubTasks,
TypeSerializerFactory<?> serializer,
TypeComparatorFactory<?> comparator,
TypePairComparatorFactory<?, ?> pairComparator) {
- AbstractJobVertex head = JobGraphUtils.createTask(IterationHeadPactTask.class, "Join With Edges (Iteration Head)", jobGraph, numSubTasks);
+ JobVertex head = JobGraphUtils.createTask(IterationHeadPactTask.class, "Join With Edges (Iteration Head)", jobGraph, numSubTasks);
TaskConfig headConfig = new TaskConfig(head.getConfiguration());
{
headConfig.setIterationId(ITERATION_ID);
@@ -292,11 +292,11 @@ public class ConnectedComponentsNepheleITCase extends RecordAPITestBase {
return head;
}
- private static AbstractJobVertex createIterationIntermediate(JobGraph jobGraph, int numSubTasks,
+ private static JobVertex createIterationIntermediate(JobGraph jobGraph, int numSubTasks,
TypeSerializerFactory<?> serializer, TypeComparatorFactory<?> comparator)
{
// --------------- the intermediate (reduce to min id) ---------------
- AbstractJobVertex intermediate = JobGraphUtils.createTask(IterationIntermediatePactTask.class,
+ JobVertex intermediate = JobGraphUtils.createTask(IterationIntermediatePactTask.class,
"Find Min Component-ID", jobGraph, numSubTasks);
TaskConfig intermediateConfig = new TaskConfig(intermediate.getConfiguration());
{
@@ -348,8 +348,8 @@ public class ConnectedComponentsNepheleITCase extends RecordAPITestBase {
return output;
}
- private static AbstractJobVertex createSync(JobGraph jobGraph, int numSubTasks, int maxIterations) {
- AbstractJobVertex sync = JobGraphUtils.createSync(jobGraph, numSubTasks);
+ private static JobVertex createSync(JobGraph jobGraph, int numSubTasks, int maxIterations) {
+ JobVertex sync = JobGraphUtils.createSync(jobGraph, numSubTasks);
TaskConfig syncConfig = new TaskConfig(sync.getConfiguration());
syncConfig.setNumberOfIterations(maxIterations);
syncConfig.setIterationId(ITERATION_ID);
@@ -380,16 +380,16 @@ public class ConnectedComponentsNepheleITCase extends RecordAPITestBase {
// -- invariant vertices -----------------------------------------------------------------------------------
InputFormatVertex vertices = createVerticesInput(jobGraph, verticesPath, numSubTasks, serializer, comparator);
InputFormatVertex edges = createEdgesInput(jobGraph, edgesPath, numSubTasks, serializer, comparator);
- AbstractJobVertex head = createIterationHead(jobGraph, numSubTasks, serializer, comparator, pairComparator);
+ JobVertex head = createIterationHead(jobGraph, numSubTasks, serializer, comparator, pairComparator);
- AbstractJobVertex intermediate = createIterationIntermediate(jobGraph, numSubTasks, serializer, comparator);
+ JobVertex intermediate = createIterationIntermediate(jobGraph, numSubTasks, serializer, comparator);
TaskConfig intermediateConfig = new TaskConfig(intermediate.getConfiguration());
OutputFormatVertex output = createOutput(jobGraph, resultPath, numSubTasks, serializer);
- AbstractJobVertex sync = createSync(jobGraph, numSubTasks, maxIterations);
+ JobVertex sync = createSync(jobGraph, numSubTasks, maxIterations);
// --------------- the tail (solution set join) ---------------
- AbstractJobVertex tail = JobGraphUtils.createTask(IterationTailPactTask.class, "IterationTail", jobGraph, numSubTasks);
+ JobVertex tail = JobGraphUtils.createTask(IterationTailPactTask.class, "IterationTail", jobGraph, numSubTasks);
TaskConfig tailConfig = new TaskConfig(tail.getConfiguration());
{
tailConfig.setIterationId(ITERATION_ID);
@@ -463,20 +463,20 @@ public class ConnectedComponentsNepheleITCase extends RecordAPITestBase {
InputFormatVertex edges = createEdgesInput(jobGraph, edgesPath, numSubTasks, serializer, comparator);
// head
- AbstractJobVertex head = createIterationHead(jobGraph, numSubTasks, serializer, comparator, pairComparator);
+ JobVertex head = createIterationHead(jobGraph, numSubTasks, serializer, comparator, pairComparator);
TaskConfig headConfig = new TaskConfig(head.getConfiguration());
headConfig.setWaitForSolutionSetUpdate();
// intermediate
- AbstractJobVertex intermediate = createIterationIntermediate(jobGraph, numSubTasks, serializer, comparator);
+ JobVertex intermediate = createIterationIntermediate(jobGraph, numSubTasks, serializer, comparator);
TaskConfig intermediateConfig = new TaskConfig(intermediate.getConfiguration());
// output and auxiliaries
OutputFormatVertex output = createOutput(jobGraph, resultPath, numSubTasks, serializer);
- AbstractJobVertex sync = createSync(jobGraph, numSubTasks, maxIterations);
+ JobVertex sync = createSync(jobGraph, numSubTasks, maxIterations);
// ------------------ the intermediate (ss join) ----------------------
- AbstractJobVertex ssJoinIntermediate = JobGraphUtils.createTask(IterationIntermediatePactTask.class,
+ JobVertex ssJoinIntermediate = JobGraphUtils.createTask(IterationIntermediatePactTask.class,
"Solution Set Join", jobGraph, numSubTasks);
TaskConfig ssJoinIntermediateConfig = new TaskConfig(ssJoinIntermediate.getConfiguration());
{
@@ -505,7 +505,7 @@ public class ConnectedComponentsNepheleITCase extends RecordAPITestBase {
}
// -------------------------- ss tail --------------------------------
- AbstractJobVertex ssTail = JobGraphUtils.createTask(IterationTailPactTask.class, "IterationSolutionSetTail",
+ JobVertex ssTail = JobGraphUtils.createTask(IterationTailPactTask.class, "IterationSolutionSetTail",
jobGraph, numSubTasks);
TaskConfig ssTailConfig = new TaskConfig(ssTail.getConfiguration());
{
@@ -529,7 +529,7 @@ public class ConnectedComponentsNepheleITCase extends RecordAPITestBase {
}
// -------------------------- ws tail --------------------------------
- AbstractJobVertex wsTail = JobGraphUtils.createTask(IterationTailPactTask.class, "IterationWorksetTail",
+ JobVertex wsTail = JobGraphUtils.createTask(IterationTailPactTask.class, "IterationWorksetTail",
jobGraph, numSubTasks);
TaskConfig wsTailConfig = new TaskConfig(wsTail.getConfiguration());
{
@@ -608,20 +608,20 @@ public class ConnectedComponentsNepheleITCase extends RecordAPITestBase {
InputFormatVertex edges = createEdgesInput(jobGraph, edgesPath, numSubTasks, serializer, comparator);
// head
- AbstractJobVertex head = createIterationHead(jobGraph, numSubTasks, serializer, comparator, pairComparator);
+ JobVertex head = createIterationHead(jobGraph, numSubTasks, serializer, comparator, pairComparator);
TaskConfig headConfig = new TaskConfig(head.getConfiguration());
headConfig.setWaitForSolutionSetUpdate();
// intermediate
- AbstractJobVertex intermediate = createIterationIntermediate(jobGraph, numSubTasks, serializer, comparator);
+ JobVertex intermediate = createIterationIntermediate(jobGraph, numSubTasks, serializer, comparator);
TaskConfig intermediateConfig = new TaskConfig(intermediate.getConfiguration());
// output and auxiliaries
- AbstractJobVertex output = createOutput(jobGraph, resultPath, numSubTasks, serializer);
- AbstractJobVertex sync = createSync(jobGraph, numSubTasks, maxIterations);
+ JobVertex output = createOutput(jobGraph, resultPath, numSubTasks, serializer);
+ JobVertex sync = createSync(jobGraph, numSubTasks, maxIterations);
// ------------------ the intermediate (ws update) ----------------------
- AbstractJobVertex wsUpdateIntermediate =
+ JobVertex wsUpdateIntermediate =
JobGraphUtils.createTask(IterationIntermediatePactTask.class, "WorksetUpdate", jobGraph, numSubTasks);
TaskConfig wsUpdateConfig = new TaskConfig(wsUpdateIntermediate.getConfiguration());
{
@@ -650,7 +650,7 @@ public class ConnectedComponentsNepheleITCase extends RecordAPITestBase {
}
// -------------------------- ss tail --------------------------------
- AbstractJobVertex ssTail =
+ JobVertex ssTail =
JobGraphUtils.createTask(IterationTailPactTask.class, "IterationSolutionSetTail", jobGraph, numSubTasks);
TaskConfig ssTailConfig = new TaskConfig(ssTail.getConfiguration());
{
@@ -729,18 +729,18 @@ public class ConnectedComponentsNepheleITCase extends RecordAPITestBase {
InputFormatVertex edges = createEdgesInput(jobGraph, edgesPath, numSubTasks, serializer, comparator);
// head
- AbstractJobVertex head = createIterationHead(jobGraph, numSubTasks, serializer, comparator, pairComparator);
+ JobVertex head = createIterationHead(jobGraph, numSubTasks, serializer, comparator, pairComparator);
// intermediate
- AbstractJobVertex intermediate = createIterationIntermediate(jobGraph, numSubTasks, serializer, comparator);
+ JobVertex intermediate = createIterationIntermediate(jobGraph, numSubTasks, serializer, comparator);
TaskConfig intermediateConfig = new TaskConfig(intermediate.getConfiguration());
// output and auxiliaries
- AbstractJobVertex output = createOutput(jobGraph, resultPath, numSubTasks, serializer);
- AbstractJobVertex sync = createSync(jobGraph, numSubTasks, maxIterations);
+ JobVertex output = createOutput(jobGraph, resultPath, numSubTasks, serializer);
+ JobVertex sync = createSync(jobGraph, numSubTasks, maxIterations);
// ------------------ the intermediate (ss update) ----------------------
- AbstractJobVertex ssJoinIntermediate = JobGraphUtils.createTask(IterationIntermediatePactTask.class,
+ JobVertex ssJoinIntermediate = JobGraphUtils.createTask(IterationIntermediatePactTask.class,
"Solution Set Update", jobGraph, numSubTasks);
TaskConfig ssJoinIntermediateConfig = new TaskConfig(ssJoinIntermediate.getConfiguration());
{
@@ -768,7 +768,7 @@ public class ConnectedComponentsNepheleITCase extends RecordAPITestBase {
}
// -------------------------- ws tail --------------------------------
- AbstractJobVertex wsTail = JobGraphUtils.createTask(IterationTailPactTask.class, "IterationWorksetTail", jobGraph, numSubTasks);
+ JobVertex wsTail = JobGraphUtils.createTask(IterationTailPactTask.class, "IterationWorksetTail", jobGraph, numSubTasks);
TaskConfig wsTailConfig = new TaskConfig(wsTail.getConfiguration());
{
wsTailConfig.setIterationId(ITERATION_ID);
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/IterationWithChainingNepheleITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/IterationWithChainingNepheleITCase.java b/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/IterationWithChainingNepheleITCase.java
index 69ff083..7a3135c 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/IterationWithChainingNepheleITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/IterationWithChainingNepheleITCase.java
@@ -31,7 +31,7 @@ import org.apache.flink.api.java.record.io.FileOutputFormat;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.iterative.task.IterationHeadPactTask;
import org.apache.flink.runtime.iterative.task.IterationTailPactTask;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.InputFormatVertex;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -138,7 +138,7 @@ public class IterationWithChainingNepheleITCase extends RecordAPITestBase {
}
// - head ------------------------------------------------------------------------------------------------------
- AbstractJobVertex head = JobGraphUtils.createTask(IterationHeadPactTask.class, "Iteration Head", jobGraph, numSubTasks);
+ JobVertex head = JobGraphUtils.createTask(IterationHeadPactTask.class, "Iteration Head", jobGraph, numSubTasks);
TaskConfig headConfig = new TaskConfig(head.getConfiguration());
{
headConfig.setIterationId(ITERATION_ID);
@@ -173,7 +173,7 @@ public class IterationWithChainingNepheleITCase extends RecordAPITestBase {
}
// - tail ------------------------------------------------------------------------------------------------------
- AbstractJobVertex tail = JobGraphUtils.createTask(IterationTailPactTask.class, "Chained Iteration Tail", jobGraph, numSubTasks);
+ JobVertex tail = JobGraphUtils.createTask(IterationTailPactTask.class, "Chained Iteration Tail", jobGraph, numSubTasks);
TaskConfig tailConfig = new TaskConfig(tail.getConfiguration());
{
tailConfig.setIterationId(ITERATION_ID);
@@ -220,7 +220,7 @@ public class IterationWithChainingNepheleITCase extends RecordAPITestBase {
}
// - sync ------------------------------------------------------------------------------------------------------
- AbstractJobVertex sync = JobGraphUtils.createSync(jobGraph, numSubTasks);
+ JobVertex sync = JobGraphUtils.createSync(jobGraph, numSubTasks);
TaskConfig syncConfig = new TaskConfig(sync.getConfiguration());
syncConfig.setNumberOfIterations(maxIterations);
syncConfig.setIterationId(ITERATION_ID);
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/JobGraphUtils.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/JobGraphUtils.java b/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/JobGraphUtils.java
index 153a85e..4edc83e 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/JobGraphUtils.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/JobGraphUtils.java
@@ -23,7 +23,7 @@ import org.apache.flink.api.common.io.InputFormat;
import org.apache.flink.api.common.operators.util.UserCodeObjectWrapper;
import org.apache.flink.api.common.operators.util.UserCodeWrapper;
import org.apache.flink.runtime.iterative.task.IterationSynchronizationSinkTask;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.InputFormatVertex;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -68,14 +68,14 @@ public class JobGraphUtils {
// new TaskConfig(source.getConfiguration()).addOutputShipStrategy(shipStrategy);
// }
- public static void connect(AbstractJobVertex source, AbstractJobVertex target, DistributionPattern distributionPattern) {
+ public static void connect(JobVertex source, JobVertex target, DistributionPattern distributionPattern) {
target.connectNewDataSetAsInput(source, distributionPattern);
}
@SuppressWarnings("rawtypes")
- public static AbstractJobVertex createTask(Class<? extends RegularPactTask> task, String name, JobGraph graph, int parallelism)
+ public static JobVertex createTask(Class<? extends RegularPactTask> task, String name, JobGraph graph, int parallelism)
{
- AbstractJobVertex taskVertex = new AbstractJobVertex(name);
+ JobVertex taskVertex = new JobVertex(name);
graph.addVertex(taskVertex);
taskVertex.setInvokableClass(task);
@@ -83,8 +83,8 @@ public class JobGraphUtils {
return taskVertex;
}
- public static AbstractJobVertex createSync(JobGraph jobGraph, int parallelism) {
- AbstractJobVertex sync = new AbstractJobVertex("BulkIterationSync");
+ public static JobVertex createSync(JobGraph jobGraph, int parallelism) {
+ JobVertex sync = new JobVertex("BulkIterationSync");
jobGraph.addVertex(sync);
sync.setInvokableClass(IterationSynchronizationSinkTask.class);
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/customdanglingpagerank/CustomCompensatableDanglingPageRank.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/customdanglingpagerank/CustomCompensatableDanglingPageRank.java b/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/customdanglingpagerank/CustomCompensatableDanglingPageRank.java
index 8801dd6..c29e868 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/customdanglingpagerank/CustomCompensatableDanglingPageRank.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/customdanglingpagerank/CustomCompensatableDanglingPageRank.java
@@ -27,7 +27,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.iterative.task.IterationHeadPactTask;
import org.apache.flink.runtime.iterative.task.IterationIntermediatePactTask;
import org.apache.flink.runtime.iterative.task.IterationTailPactTask;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.InputFormatVertex;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -154,7 +154,7 @@ public class CustomCompensatableDanglingPageRank {
adjacencyListInputConfig.setOutputComparator(vertexWithAdjacencyListComparator, 0);
// --------------- the head ---------------------
- AbstractJobVertex head = JobGraphUtils.createTask(IterationHeadPactTask.class, "IterationHead", jobGraph,
+ JobVertex head = JobGraphUtils.createTask(IterationHeadPactTask.class, "IterationHead", jobGraph,
parallelism);
TaskConfig headConfig = new TaskConfig(head.getConfiguration());
headConfig.setIterationId(ITERATION_ID);
@@ -199,7 +199,7 @@ public class CustomCompensatableDanglingPageRank {
// --------------- the join ---------------------
- AbstractJobVertex intermediate = JobGraphUtils.createTask(IterationIntermediatePactTask.class,
+ JobVertex intermediate = JobGraphUtils.createTask(IterationIntermediatePactTask.class,
"IterationIntermediate", jobGraph, parallelism);
TaskConfig intermediateConfig = new TaskConfig(intermediate.getConfiguration());
intermediateConfig.setIterationId(ITERATION_ID);
@@ -227,7 +227,7 @@ public class CustomCompensatableDanglingPageRank {
// ---------------- the tail (co group) --------------------
- AbstractJobVertex tail = JobGraphUtils.createTask(IterationTailPactTask.class, "IterationTail", jobGraph,
+ JobVertex tail = JobGraphUtils.createTask(IterationTailPactTask.class, "IterationTail", jobGraph,
parallelism);
TaskConfig tailConfig = new TaskConfig(tail.getConfiguration());
tailConfig.setIterationId(ITERATION_ID);
@@ -273,7 +273,7 @@ public class CustomCompensatableDanglingPageRank {
// --------------- the auxiliaries ---------------------
- AbstractJobVertex sync = JobGraphUtils.createSync(jobGraph, parallelism);
+ JobVertex sync = JobGraphUtils.createSync(jobGraph, parallelism);
TaskConfig syncConfig = new TaskConfig(sync.getConfiguration());
syncConfig.setNumberOfIterations(numIterations);
syncConfig.addIterationAggregator(CustomCompensatableDotProductCoGroup.AGGREGATOR_NAME, new PageRankStatsAggregator());
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/customdanglingpagerank/CustomCompensatableDanglingPageRankWithCombiner.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/customdanglingpagerank/CustomCompensatableDanglingPageRankWithCombiner.java b/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/customdanglingpagerank/CustomCompensatableDanglingPageRankWithCombiner.java
index 6f19c03..de9e0a1 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/customdanglingpagerank/CustomCompensatableDanglingPageRankWithCombiner.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/customdanglingpagerank/CustomCompensatableDanglingPageRankWithCombiner.java
@@ -27,7 +27,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.iterative.task.IterationHeadPactTask;
import org.apache.flink.runtime.iterative.task.IterationIntermediatePactTask;
import org.apache.flink.runtime.iterative.task.IterationTailPactTask;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.InputFormatVertex;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -154,7 +154,7 @@ public class CustomCompensatableDanglingPageRankWithCombiner {
adjacencyListInputConfig.setOutputComparator(vertexWithAdjacencyListComparator, 0);
// --------------- the head ---------------------
- AbstractJobVertex head = JobGraphUtils.createTask(IterationHeadPactTask.class, "IterationHead", jobGraph,
+ JobVertex head = JobGraphUtils.createTask(IterationHeadPactTask.class, "IterationHead", jobGraph,
parallelism);
TaskConfig headConfig = new TaskConfig(head.getConfiguration());
headConfig.setIterationId(ITERATION_ID);
@@ -199,7 +199,7 @@ public class CustomCompensatableDanglingPageRankWithCombiner {
// --------------- the join ---------------------
- AbstractJobVertex intermediate = JobGraphUtils.createTask(IterationIntermediatePactTask.class,
+ JobVertex intermediate = JobGraphUtils.createTask(IterationIntermediatePactTask.class,
"IterationIntermediate", jobGraph, parallelism);
TaskConfig intermediateConfig = new TaskConfig(intermediate.getConfiguration());
intermediateConfig.setIterationId(ITERATION_ID);
@@ -240,7 +240,7 @@ public class CustomCompensatableDanglingPageRankWithCombiner {
// ---------------- the tail (co group) --------------------
- AbstractJobVertex tail = JobGraphUtils.createTask(IterationTailPactTask.class, "IterationTail", jobGraph,
+ JobVertex tail = JobGraphUtils.createTask(IterationTailPactTask.class, "IterationTail", jobGraph,
parallelism);
TaskConfig tailConfig = new TaskConfig(tail.getConfiguration());
tailConfig.setIterationId(ITERATION_ID);
@@ -287,7 +287,7 @@ public class CustomCompensatableDanglingPageRankWithCombiner {
// --------------- the auxiliaries ---------------------
- AbstractJobVertex sync = JobGraphUtils.createSync(jobGraph, parallelism);
+ JobVertex sync = JobGraphUtils.createSync(jobGraph, parallelism);
TaskConfig syncConfig = new TaskConfig(sync.getConfiguration());
syncConfig.setNumberOfIterations(numIterations);
syncConfig.addIterationAggregator(CustomCompensatableDotProductCoGroup.AGGREGATOR_NAME, new PageRankStatsAggregator());
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/danglingpagerank/CompensatableDanglingPageRank.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/danglingpagerank/CompensatableDanglingPageRank.java b/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/danglingpagerank/CompensatableDanglingPageRank.java
index 8216ccb..eb2ccdc 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/danglingpagerank/CompensatableDanglingPageRank.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/iterative/nephele/danglingpagerank/CompensatableDanglingPageRank.java
@@ -31,7 +31,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.iterative.task.IterationHeadPactTask;
import org.apache.flink.runtime.iterative.task.IterationIntermediatePactTask;
import org.apache.flink.runtime.iterative.task.IterationTailPactTask;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.InputFormatVertex;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -135,7 +135,7 @@ public class CompensatableDanglingPageRank {
adjacencyListInputConfig.setOutputComparator(fieldZeroComparator, 0);
// --------------- the head ---------------------
- AbstractJobVertex head = JobGraphUtils.createTask(IterationHeadPactTask.class, "IterationHead", jobGraph, parallelism);
+ JobVertex head = JobGraphUtils.createTask(IterationHeadPactTask.class, "IterationHead", jobGraph, parallelism);
TaskConfig headConfig = new TaskConfig(head.getConfiguration());
headConfig.setIterationId(ITERATION_ID);
@@ -179,7 +179,7 @@ public class CompensatableDanglingPageRank {
// --------------- the join ---------------------
- AbstractJobVertex intermediate = JobGraphUtils.createTask(IterationIntermediatePactTask.class, "IterationIntermediate", jobGraph, parallelism);
+ JobVertex intermediate = JobGraphUtils.createTask(IterationIntermediatePactTask.class, "IterationIntermediate", jobGraph, parallelism);
TaskConfig intermediateConfig = new TaskConfig(intermediate.getConfiguration());
intermediateConfig.setIterationId(ITERATION_ID);
// intermediateConfig.setDriver(RepeatableHashjoinMatchDriverWithCachedBuildside.class);
@@ -206,7 +206,7 @@ public class CompensatableDanglingPageRank {
// ---------------- the tail (co group) --------------------
- AbstractJobVertex tail = JobGraphUtils.createTask(IterationTailPactTask.class, "IterationTail", jobGraph,
+ JobVertex tail = JobGraphUtils.createTask(IterationTailPactTask.class, "IterationTail", jobGraph,
parallelism);
TaskConfig tailConfig = new TaskConfig(tail.getConfiguration());
tailConfig.setIterationId(ITERATION_ID);
@@ -253,7 +253,7 @@ public class CompensatableDanglingPageRank {
// --------------- the auxiliaries ---------------------
- AbstractJobVertex sync = JobGraphUtils.createSync(jobGraph, parallelism);
+ JobVertex sync = JobGraphUtils.createSync(jobGraph, parallelism);
TaskConfig syncConfig = new TaskConfig(sync.getConfiguration());
syncConfig.setNumberOfIterations(numIterations);
syncConfig.addIterationAggregator(CompensatableDotProductCoGroup.AGGREGATOR_NAME, new PageRankStatsAggregator());
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-tests/src/test/java/org/apache/flink/test/runtime/NetworkStackThroughputITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/runtime/NetworkStackThroughputITCase.java b/flink-tests/src/test/java/org/apache/flink/test/runtime/NetworkStackThroughputITCase.java
index 7b43266..310ded8 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/runtime/NetworkStackThroughputITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/runtime/NetworkStackThroughputITCase.java
@@ -24,7 +24,7 @@ import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.runtime.io.network.api.reader.RecordReader;
import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
@@ -100,7 +100,7 @@ public class NetworkStackThroughputITCase {
JobGraph jobGraph = new JobGraph("Speed Test");
SlotSharingGroup sharingGroup = new SlotSharingGroup();
- AbstractJobVertex producer = new AbstractJobVertex("Speed Test Producer");
+ JobVertex producer = new JobVertex("Speed Test Producer");
jobGraph.addVertex(producer);
producer.setSlotSharingGroup(sharingGroup);
@@ -109,9 +109,9 @@ public class NetworkStackThroughputITCase {
producer.getConfiguration().setInteger(DATA_VOLUME_GB_CONFIG_KEY, dataVolumeGb);
producer.getConfiguration().setBoolean(IS_SLOW_SENDER_CONFIG_KEY, isSlowSender);
- AbstractJobVertex forwarder = null;
+ JobVertex forwarder = null;
if (useForwarder) {
- forwarder = new AbstractJobVertex("Speed Test Forwarder");
+ forwarder = new JobVertex("Speed Test Forwarder");
jobGraph.addVertex(forwarder);
forwarder.setSlotSharingGroup(sharingGroup);
@@ -119,7 +119,7 @@ public class NetworkStackThroughputITCase {
forwarder.setParallelism(numSubtasks);
}
- AbstractJobVertex consumer = new AbstractJobVertex("Speed Test Consumer");
+ JobVertex consumer = new JobVertex("Speed Test Consumer");
jobGraph.addVertex(consumer);
consumer.setSlotSharingGroup(sharingGroup);
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerFailsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerFailsITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerFailsITCase.scala
index ca8dd75..a590ae7 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerFailsITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerFailsITCase.scala
@@ -27,7 +27,7 @@ import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}
import org.apache.flink.configuration.{ConfigConstants, Configuration}
import org.apache.flink.runtime.akka.AkkaUtils
-import org.apache.flink.runtime.jobgraph.{AbstractJobVertex, JobGraph}
+import org.apache.flink.runtime.jobgraph.{JobVertex, JobGraph}
import org.apache.flink.runtime.jobmanager.Tasks.{BlockingNoOpInvokable, NoOpInvokable}
import org.apache.flink.runtime.messages.JobManagerMessages._
import org.apache.flink.runtime.testingUtils.TestingMessages.DisableDisconnect
@@ -82,12 +82,12 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
"go into a clean state in case of a JobManager failure" in {
val num_slots = 36
- val sender = new AbstractJobVertex("BlockingSender")
+ val sender = new JobVertex("BlockingSender")
sender.setParallelism(num_slots)
sender.setInvokableClass(classOf[BlockingNoOpInvokable])
val jobGraph = new JobGraph("Blocking Testjob", sender)
- val noOp = new AbstractJobVertex("NoOpInvokable")
+ val noOp = new JobVertex("NoOpInvokable")
noOp.setParallelism(num_slots)
noOp.setInvokableClass(classOf[NoOpInvokable])
val jobGraph2 = new JobGraph("NoOp Testjob", noOp)
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/taskmanager/TaskManagerFailsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/taskmanager/TaskManagerFailsITCase.scala b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/taskmanager/TaskManagerFailsITCase.scala
index 07eaed7..1884fab 100644
--- a/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/taskmanager/TaskManagerFailsITCase.scala
+++ b/flink-tests/src/test/scala/org/apache/flink/api/scala/runtime/taskmanager/TaskManagerFailsITCase.scala
@@ -26,7 +26,7 @@ import org.apache.flink.configuration.ConfigConstants
import org.apache.flink.configuration.Configuration
import org.apache.flink.runtime.akka.AkkaUtils
import org.apache.flink.runtime.client.JobExecutionException
-import org.apache.flink.runtime.jobgraph.{AbstractJobVertex, DistributionPattern, JobGraph}
+import org.apache.flink.runtime.jobgraph.{JobVertex, DistributionPattern, JobGraph}
import org.apache.flink.runtime.jobmanager.Tasks.{NoOpInvokable, BlockingNoOpInvokable, BlockingReceiver, Sender}
import org.apache.flink.runtime.messages.JobManagerMessages.{JobResultSuccess, RequestNumberRegisteredTaskManager, SubmitJob}
import org.apache.flink.runtime.messages.TaskManagerMessages.{RegisteredAtJobManager, NotifyWhenRegisteredAtJobManager}
@@ -84,8 +84,8 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
"handle gracefully failing task manager" in {
val num_tasks = 31
- val sender = new AbstractJobVertex("Sender")
- val receiver = new AbstractJobVertex("Receiver")
+ val sender = new JobVertex("Sender")
+ val receiver = new JobVertex("Receiver")
sender.setInvokableClass(classOf[Sender])
receiver.setInvokableClass(classOf[BlockingReceiver])
sender.setParallelism(num_tasks)
@@ -130,8 +130,8 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
"handle hard failing task manager" in {
val num_tasks = 31
- val sender = new AbstractJobVertex("Sender")
- val receiver = new AbstractJobVertex("Receiver")
+ val sender = new JobVertex("Sender")
+ val receiver = new JobVertex("Receiver")
sender.setInvokableClass(classOf[Sender])
receiver.setInvokableClass(classOf[BlockingReceiver])
sender.setParallelism(num_tasks)
@@ -174,12 +174,12 @@ with WordSpecLike with Matchers with BeforeAndAfterAll {
"go into a clean state in case of a TaskManager failure" in {
val num_slots = 20
- val sender = new AbstractJobVertex("BlockingSender")
+ val sender = new JobVertex("BlockingSender")
sender.setParallelism(num_slots)
sender.setInvokableClass(classOf[BlockingNoOpInvokable])
val jobGraph = new JobGraph("Blocking Testjob", sender)
- val noOp = new AbstractJobVertex("NoOpInvokable")
+ val noOp = new JobVertex("NoOpInvokable")
noOp.setParallelism(num_slots)
noOp.setInvokableClass(classOf[NoOpInvokable])
val jobGraph2 = new JobGraph("NoOp Testjob", noOp)
[2/3] flink git commit: [FLINK-2120][runtime] rename
AbstractJobVertex to JobVertex
Posted by mx...@apache.org.
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/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 894a7a9..a4bd03c 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
@@ -44,7 +44,7 @@ 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.AbstractJobVertex;
+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;
@@ -79,11 +79,11 @@ public class ExecutionGraphConstructionTest {
final String jobName = "Test Job Sample Name";
final Configuration cfg = new Configuration();
- AbstractJobVertex v1 = new AbstractJobVertex("vertex1");
- AbstractJobVertex v2 = new AbstractJobVertex("vertex2");
- AbstractJobVertex v3 = new AbstractJobVertex("vertex3");
- AbstractJobVertex v4 = new AbstractJobVertex("vertex4");
- AbstractJobVertex v5 = new AbstractJobVertex("vertex5");
+ 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");
v1.setParallelism(5);
v2.setParallelism(7);
@@ -97,7 +97,7 @@ public class ExecutionGraphConstructionTest {
v5.connectNewDataSetAsInput(v4, DistributionPattern.ALL_TO_ALL);
v5.connectNewDataSetAsInput(v3, DistributionPattern.ALL_TO_ALL);
- List<AbstractJobVertex> ordered = new ArrayList<AbstractJobVertex>(Arrays.asList(v1, v2, v3, v4, v5));
+ List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2, v3, v4, v5));
ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
try {
@@ -118,9 +118,9 @@ public class ExecutionGraphConstructionTest {
final Configuration cfg = new Configuration();
// construct part one of the execution graph
- AbstractJobVertex v1 = new AbstractJobVertex("vertex1");
- AbstractJobVertex v2 = new AbstractJobVertex("vertex2");
- AbstractJobVertex v3 = new AbstractJobVertex("vertex3");
+ JobVertex v1 = new JobVertex("vertex1");
+ JobVertex v2 = new JobVertex("vertex2");
+ JobVertex v3 = new JobVertex("vertex3");
v1.setParallelism(5);
v2.setParallelism(7);
@@ -135,7 +135,7 @@ public class ExecutionGraphConstructionTest {
IntermediateDataSet v3result_2 = v3.createAndAddResultDataSet(ResultPartitionType.PIPELINED);
- List<AbstractJobVertex> ordered = new ArrayList<AbstractJobVertex>(Arrays.asList(v1, v2, v3));
+ List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2, v3));
ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
try {
@@ -148,8 +148,8 @@ public class ExecutionGraphConstructionTest {
// attach the second part of the graph
- AbstractJobVertex v4 = new AbstractJobVertex("vertex4");
- AbstractJobVertex v5 = new AbstractJobVertex("vertex5");
+ JobVertex v4 = new JobVertex("vertex4");
+ JobVertex v5 = new JobVertex("vertex5");
v4.setParallelism(11);
v5.setParallelism(4);
@@ -158,7 +158,7 @@ public class ExecutionGraphConstructionTest {
v5.connectNewDataSetAsInput(v4, DistributionPattern.ALL_TO_ALL);
v5.connectDataSetAsInput(v3result_2, DistributionPattern.ALL_TO_ALL);
- List<AbstractJobVertex> ordered2 = new ArrayList<AbstractJobVertex>(Arrays.asList(v4, v5));
+ List<JobVertex> ordered2 = new ArrayList<JobVertex>(Arrays.asList(v4, v5));
try {
eg.attachJobGraph(ordered2);
@@ -179,9 +179,9 @@ public class ExecutionGraphConstructionTest {
final Configuration cfg = new Configuration();
// construct part one of the execution graph
- AbstractJobVertex v1 = new AbstractJobVertex("vertex1");
- AbstractJobVertex v2 = new AbstractJobVertex("vertex2");
- AbstractJobVertex v3 = new AbstractJobVertex("vertex3");
+ JobVertex v1 = new JobVertex("vertex1");
+ JobVertex v2 = new JobVertex("vertex2");
+ JobVertex v3 = new JobVertex("vertex3");
v1.setParallelism(5);
v2.setParallelism(7);
@@ -196,7 +196,7 @@ public class ExecutionGraphConstructionTest {
IntermediateDataSet v3result_2 = v3.createAndAddResultDataSet(ResultPartitionType.PIPELINED);
- List<AbstractJobVertex> ordered = new ArrayList<AbstractJobVertex>(Arrays.asList(v1, v2, v3));
+ List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2, v3));
ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
try {
@@ -209,8 +209,8 @@ public class ExecutionGraphConstructionTest {
// attach the second part of the graph
- AbstractJobVertex v4 = new AbstractJobVertex("vertex4");
- AbstractJobVertex v5 = new AbstractJobVertex("vertex5");
+ JobVertex v4 = new JobVertex("vertex4");
+ JobVertex v5 = new JobVertex("vertex5");
v4.setParallelism(11);
v5.setParallelism(4);
@@ -219,7 +219,7 @@ public class ExecutionGraphConstructionTest {
v5.connectNewDataSetAsInput(v4, DistributionPattern.ALL_TO_ALL);
v5.connectIdInput(v3result_2.getId(), DistributionPattern.ALL_TO_ALL);
- List<AbstractJobVertex> ordered2 = new ArrayList<AbstractJobVertex>(Arrays.asList(v4, v5));
+ List<JobVertex> ordered2 = new ArrayList<JobVertex>(Arrays.asList(v4, v5));
try {
eg.attachJobGraph(ordered2);
@@ -234,8 +234,8 @@ public class ExecutionGraphConstructionTest {
}
private void verifyTestGraph(ExecutionGraph eg, JobID jobId,
- AbstractJobVertex v1, AbstractJobVertex v2, AbstractJobVertex v3,
- AbstractJobVertex v4, AbstractJobVertex v5)
+ JobVertex v1, JobVertex v2, JobVertex v3,
+ JobVertex v4, JobVertex v5)
{
Map<JobVertexID, ExecutionJobVertex> vertices = eg.getAllVertices();
@@ -441,10 +441,10 @@ public class ExecutionGraphConstructionTest {
final Configuration cfg = new Configuration();
// construct part one of the execution graph
- AbstractJobVertex v1 = new AbstractJobVertex("vertex1");
+ JobVertex v1 = new JobVertex("vertex1");
v1.setParallelism(7);
- List<AbstractJobVertex> ordered = new ArrayList<AbstractJobVertex>(Arrays.asList(v1));
+ List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1));
ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
try {
@@ -456,10 +456,10 @@ public class ExecutionGraphConstructionTest {
}
// attach the second part of the graph
- AbstractJobVertex v2 = new AbstractJobVertex("vertex2");
+ JobVertex v2 = new JobVertex("vertex2");
v2.connectIdInput(new IntermediateDataSetID(), DistributionPattern.ALL_TO_ALL);
- List<AbstractJobVertex> ordered2 = new ArrayList<AbstractJobVertex>(Arrays.asList(v2));
+ List<JobVertex> ordered2 = new ArrayList<JobVertex>(Arrays.asList(v2));
try {
eg.attachJobGraph(ordered2);
@@ -476,11 +476,11 @@ public class ExecutionGraphConstructionTest {
final String jobName = "Test Job Sample Name";
final Configuration cfg = new Configuration();
- AbstractJobVertex v1 = new AbstractJobVertex("vertex1");
- AbstractJobVertex v2 = new AbstractJobVertex("vertex2");
- AbstractJobVertex v3 = new AbstractJobVertex("vertex3");
- AbstractJobVertex v4 = new AbstractJobVertex("vertex4");
- AbstractJobVertex v5 = new AbstractJobVertex("vertex5");
+ 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");
v1.setParallelism(5);
v2.setParallelism(7);
@@ -494,7 +494,7 @@ public class ExecutionGraphConstructionTest {
v5.connectNewDataSetAsInput(v4, DistributionPattern.ALL_TO_ALL);
v5.connectNewDataSetAsInput(v3, DistributionPattern.ALL_TO_ALL);
- List<AbstractJobVertex> ordered = new ArrayList<AbstractJobVertex>(Arrays.asList(v1, v2, v3, v5, v4));
+ List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2, v3, v5, v4));
ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
try {
@@ -528,11 +528,11 @@ public class ExecutionGraphConstructionTest {
final String jobName = "Test Job Sample Name";
final Configuration cfg = new Configuration();
- AbstractJobVertex v1 = new AbstractJobVertex("vertex1");
- AbstractJobVertex v2 = new AbstractJobVertex("vertex2");
- AbstractJobVertex v3 = new AbstractJobVertex("vertex3");
- AbstractJobVertex v4 = new AbstractJobVertex("vertex4");
- AbstractJobVertex v5 = new AbstractJobVertex("vertex5");
+ 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");
v1.setParallelism(5);
v2.setParallelism(7);
@@ -549,7 +549,7 @@ public class ExecutionGraphConstructionTest {
v3.setInputSplitSource(source1);
v5.setInputSplitSource(source2);
- List<AbstractJobVertex> ordered = new ArrayList<AbstractJobVertex>(Arrays.asList(v1, v2, v3, v4, v5));
+ List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2, v3, v4, v5));
ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg,
AkkaUtils.getDefaultTimeout());
@@ -577,9 +577,9 @@ public class ExecutionGraphConstructionTest {
final String jobName = "Test Job Sample Name";
final Configuration cfg = new Configuration();
- AbstractJobVertex v1 = new AbstractJobVertex("vertex1");
- AbstractJobVertex v2 = new AbstractJobVertex("vertex2");
- AbstractJobVertex v3 = new AbstractJobVertex("vertex3");
+ JobVertex v1 = new JobVertex("vertex1");
+ JobVertex v2 = new JobVertex("vertex2");
+ JobVertex v3 = new JobVertex("vertex3");
v1.setParallelism(5);
v2.setParallelism(7);
@@ -589,7 +589,7 @@ public class ExecutionGraphConstructionTest {
v2.connectDataSetAsInput(result, DistributionPattern.ALL_TO_ALL);
v3.connectDataSetAsInput(result, DistributionPattern.ALL_TO_ALL);
- List<AbstractJobVertex> ordered = new ArrayList<AbstractJobVertex>(Arrays.asList(v1, v2, v3));
+ List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2, v3));
ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg,
AkkaUtils.getDefaultTimeout());
@@ -616,8 +616,8 @@ public class ExecutionGraphConstructionTest {
final Configuration cfg = new Configuration();
// simple group of two, cyclic
- AbstractJobVertex v1 = new AbstractJobVertex("vertex1");
- AbstractJobVertex v2 = new AbstractJobVertex("vertex2");
+ JobVertex v1 = new JobVertex("vertex1");
+ JobVertex v2 = new JobVertex("vertex2");
v1.setParallelism(6);
v2.setParallelism(4);
@@ -628,11 +628,11 @@ public class ExecutionGraphConstructionTest {
v1.setStrictlyCoLocatedWith(v2);
// complex forked dependency pattern
- AbstractJobVertex v3 = new AbstractJobVertex("vertex3");
- AbstractJobVertex v4 = new AbstractJobVertex("vertex4");
- AbstractJobVertex v5 = new AbstractJobVertex("vertex5");
- AbstractJobVertex v6 = new AbstractJobVertex("vertex6");
- AbstractJobVertex v7 = new AbstractJobVertex("vertex7");
+ JobVertex v3 = new JobVertex("vertex3");
+ JobVertex v4 = new JobVertex("vertex4");
+ JobVertex v5 = new JobVertex("vertex5");
+ JobVertex v6 = new JobVertex("vertex6");
+ JobVertex v7 = new JobVertex("vertex7");
v3.setParallelism(3);
v4.setParallelism(3);
v5.setParallelism(3);
@@ -652,7 +652,7 @@ public class ExecutionGraphConstructionTest {
v3.setStrictlyCoLocatedWith(v7);
// isolated vertex
- AbstractJobVertex v8 = new AbstractJobVertex("vertex8");
+ JobVertex v8 = new JobVertex("vertex8");
v8.setParallelism(2);
JobGraph jg = new JobGraph(jobId, jobName, v1, v2, v3, v4, v5, v6, v7, v8);
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java
index a53c318..03a41b4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphDeploymentTest.java
@@ -44,7 +44,7 @@ import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.instance.SimpleSlot;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.jobgraph.JobVertexID;
@@ -81,10 +81,10 @@ public class ExecutionGraphDeploymentTest {
final JobVertexID jid3 = new JobVertexID();
final JobVertexID jid4 = new JobVertexID();
- AbstractJobVertex v1 = new AbstractJobVertex("v1", jid1);
- AbstractJobVertex v2 = new AbstractJobVertex("v2", jid2);
- AbstractJobVertex v3 = new AbstractJobVertex("v3", jid3);
- AbstractJobVertex v4 = new AbstractJobVertex("v4", jid4);
+ JobVertex v1 = new JobVertex("v1", jid1);
+ JobVertex v2 = new JobVertex("v2", jid2);
+ JobVertex v3 = new JobVertex("v3", jid3);
+ JobVertex v4 = new JobVertex("v4", jid4);
v1.setParallelism(10);
v2.setParallelism(10);
@@ -103,7 +103,7 @@ public class ExecutionGraphDeploymentTest {
ExecutionGraph eg = new ExecutionGraph(jobId, "some job", new Configuration(),
AkkaUtils.getDefaultTimeout());
- List<AbstractJobVertex> ordered = Arrays.asList(v1, v2, v3, v4);
+ List<JobVertex> ordered = Arrays.asList(v1, v2, v3, v4);
eg.attachJobGraph(ordered);
@@ -163,8 +163,8 @@ public class ExecutionGraphDeploymentTest {
final JobVertexID jid1 = new JobVertexID();
final JobVertexID jid2 = new JobVertexID();
- AbstractJobVertex v1 = new AbstractJobVertex("v1", jid1);
- AbstractJobVertex v2 = new AbstractJobVertex("v2", jid2);
+ JobVertex v1 = new JobVertex("v1", jid1);
+ JobVertex v2 = new JobVertex("v2", jid2);
Map<ExecutionAttemptID, Execution> executions = setupExecution(v1, 7650, v2, 2350);
@@ -187,8 +187,8 @@ public class ExecutionGraphDeploymentTest {
final JobVertexID jid1 = new JobVertexID();
final JobVertexID jid2 = new JobVertexID();
- AbstractJobVertex v1 = new AbstractJobVertex("v1", jid1);
- AbstractJobVertex v2 = new AbstractJobVertex("v2", jid2);
+ JobVertex v1 = new JobVertex("v1", jid1);
+ JobVertex v2 = new JobVertex("v2", jid2);
Map<ExecutionAttemptID, Execution> executions = setupExecution(v1, 7, v2, 6);
@@ -211,8 +211,8 @@ public class ExecutionGraphDeploymentTest {
final JobVertexID jid1 = new JobVertexID();
final JobVertexID jid2 = new JobVertexID();
- AbstractJobVertex v1 = new AbstractJobVertex("v1", jid1);
- AbstractJobVertex v2 = new AbstractJobVertex("v2", jid2);
+ JobVertex v1 = new JobVertex("v1", jid1);
+ JobVertex v2 = new JobVertex("v2", jid2);
Map<ExecutionAttemptID, Execution> executions = setupExecution(v1, 7, v2, 6);
@@ -235,8 +235,8 @@ public class ExecutionGraphDeploymentTest {
final JobVertexID jid1 = new JobVertexID();
final JobVertexID jid2 = new JobVertexID();
- AbstractJobVertex v1 = new AbstractJobVertex("v1", jid1);
- AbstractJobVertex v2 = new AbstractJobVertex("v2", jid2);
+ JobVertex v1 = new JobVertex("v1", jid1);
+ JobVertex v2 = new JobVertex("v2", jid2);
Map<ExecutionAttemptID, Execution> executions = setupExecution(v1, 19, v2, 37);
@@ -260,8 +260,8 @@ public class ExecutionGraphDeploymentTest {
final JobVertexID jid1 = new JobVertexID();
final JobVertexID jid2 = new JobVertexID();
- AbstractJobVertex v1 = new FailingFinalizeJobVertex("v1", jid1);
- AbstractJobVertex v2 = new AbstractJobVertex("v2", jid2);
+ JobVertex v1 = new FailingFinalizeJobVertex("v1", jid1);
+ JobVertex v2 = new JobVertex("v2", jid2);
Map<ExecutionAttemptID, Execution> executions = setupExecution(v1, 6, v2, 4);
@@ -297,7 +297,7 @@ public class ExecutionGraphDeploymentTest {
}
}
- private Map<ExecutionAttemptID, Execution> setupExecution(AbstractJobVertex v1, int dop1, AbstractJobVertex v2, int dop2) throws Exception {
+ private Map<ExecutionAttemptID, Execution> setupExecution(JobVertex v1, int dop1, JobVertex v2, int dop2) throws Exception {
final JobID jobId = new JobID();
v1.setParallelism(dop1);
@@ -311,7 +311,7 @@ public class ExecutionGraphDeploymentTest {
AkkaUtils.getDefaultTimeout());
eg.setQueuedSchedulingAllowed(false);
- List<AbstractJobVertex> ordered = Arrays.asList(v1, v2);
+ List<JobVertex> ordered = Arrays.asList(v1, v2);
eg.attachJobGraph(ordered);
// create a mock taskmanager that accepts deployment calls
@@ -333,7 +333,7 @@ public class ExecutionGraphDeploymentTest {
}
@SuppressWarnings("serial")
- public static class FailingFinalizeJobVertex extends AbstractJobVertex {
+ public static class FailingFinalizeJobVertex extends JobVertex {
public FailingFinalizeJobVertex(String name) {
super(name);
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/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 a1ee79b..a77a09e 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
@@ -39,7 +39,7 @@ import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.instance.InstanceConnectionInfo;
import org.apache.flink.runtime.instance.InstanceID;
import org.apache.flink.runtime.instance.SimpleSlot;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.jobgraph.JobVertexID;
@@ -147,7 +147,7 @@ public class ExecutionGraphTestUtils {
}
public static ExecutionJobVertex getExecutionVertex(JobVertexID id) throws JobException {
- AbstractJobVertex ajv = new AbstractJobVertex("TestVertex", id);
+ JobVertex ajv = new JobVertex("TestVertex", id);
ajv.setInvokableClass(mock(AbstractInvokable.class).getClass());
ExecutionGraph graph = new ExecutionGraph(new JobID(), "test job", new Configuration(),
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java
index 733ad11..7787ab4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java
@@ -31,7 +31,7 @@ import akka.testkit.JavaTestKit;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.instance.SimpleSlot;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.jobgraph.JobVertexID;
@@ -61,7 +61,7 @@ public class ExecutionStateProgressTest {
final JobID jid = new JobID();
final JobVertexID vid = new JobVertexID();
- AbstractJobVertex ajv = new AbstractJobVertex("TestVertex", vid);
+ JobVertex ajv = new JobVertex("TestVertex", vid);
ajv.setParallelism(3);
ajv.setInvokableClass(mock(AbstractInvokable.class).getClass());
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/LocalInputSplitsTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/LocalInputSplitsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/LocalInputSplitsTest.java
index f0001a9..0d2ffeb 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/LocalInputSplitsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/LocalInputSplitsTest.java
@@ -34,7 +34,7 @@ import org.apache.flink.runtime.instance.HardwareDescription;
import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.instance.InstanceConnectionInfo;
import org.apache.flink.runtime.instance.InstanceID;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobmanager.scheduler.Scheduler;
import org.apache.flink.runtime.operators.testutils.DummyInvokable;
@@ -283,7 +283,7 @@ public class LocalInputSplitsTest {
};
try {
- AbstractJobVertex vertex = new AbstractJobVertex("test vertex");
+ JobVertex vertex = new JobVertex("test vertex");
vertex.setParallelism(6);
vertex.setInvokableClass(DummyInvokable.class);
vertex.setInputSplitSource(new TestInputSplitSource(splits));
@@ -342,7 +342,7 @@ public class LocalInputSplitsTest {
TestLocatableInputSplit[] splits)
throws Exception
{
- AbstractJobVertex vertex = new AbstractJobVertex("test vertex");
+ JobVertex vertex = new JobVertex("test vertex");
vertex.setParallelism(parallelism);
vertex.setInvokableClass(DummyInvokable.class);
vertex.setInputSplitSource(new TestInputSplitSource(splits));
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java
index f72d105..4677bf8 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java
@@ -31,7 +31,7 @@ import java.util.List;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.JobException;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.api.common.JobID;
@@ -46,15 +46,15 @@ public class PointwisePatternTest {
public void testNToN() {
final int N = 23;
- AbstractJobVertex v1 = new AbstractJobVertex("vertex1");
- AbstractJobVertex v2 = new AbstractJobVertex("vertex2");
+ JobVertex v1 = new JobVertex("vertex1");
+ JobVertex v2 = new JobVertex("vertex2");
v1.setParallelism(N);
v2.setParallelism(N);
v2.connectNewDataSetAsInput(v1, DistributionPattern.POINTWISE);
- List<AbstractJobVertex> ordered = new ArrayList<AbstractJobVertex>(Arrays.asList(v1, v2));
+ List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2));
ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
try {
@@ -81,15 +81,15 @@ public class PointwisePatternTest {
public void test2NToN() {
final int N = 17;
- AbstractJobVertex v1 = new AbstractJobVertex("vertex1");
- AbstractJobVertex v2 = new AbstractJobVertex("vertex2");
+ JobVertex v1 = new JobVertex("vertex1");
+ JobVertex v2 = new JobVertex("vertex2");
v1.setParallelism(2 * N);
v2.setParallelism(N);
v2.connectNewDataSetAsInput(v1, DistributionPattern.POINTWISE);
- List<AbstractJobVertex> ordered = new ArrayList<AbstractJobVertex>(Arrays.asList(v1, v2));
+ List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2));
ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
try {
@@ -117,15 +117,15 @@ public class PointwisePatternTest {
public void test3NToN() {
final int N = 17;
- AbstractJobVertex v1 = new AbstractJobVertex("vertex1");
- AbstractJobVertex v2 = new AbstractJobVertex("vertex2");
+ JobVertex v1 = new JobVertex("vertex1");
+ JobVertex v2 = new JobVertex("vertex2");
v1.setParallelism(3 * N);
v2.setParallelism(N);
v2.connectNewDataSetAsInput(v1, DistributionPattern.POINTWISE);
- List<AbstractJobVertex> ordered = new ArrayList<AbstractJobVertex>(Arrays.asList(v1, v2));
+ List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2));
ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
try {
@@ -154,15 +154,15 @@ public class PointwisePatternTest {
public void testNTo2N() {
final int N = 41;
- AbstractJobVertex v1 = new AbstractJobVertex("vertex1");
- AbstractJobVertex v2 = new AbstractJobVertex("vertex2");
+ JobVertex v1 = new JobVertex("vertex1");
+ JobVertex v2 = new JobVertex("vertex2");
v1.setParallelism(N);
v2.setParallelism(2 * N);
v2.connectNewDataSetAsInput(v1, DistributionPattern.POINTWISE);
- List<AbstractJobVertex> ordered = new ArrayList<AbstractJobVertex>(Arrays.asList(v1, v2));
+ List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2));
ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
try {
@@ -189,15 +189,15 @@ public class PointwisePatternTest {
public void testNTo7N() {
final int N = 11;
- AbstractJobVertex v1 = new AbstractJobVertex("vertex1");
- AbstractJobVertex v2 = new AbstractJobVertex("vertex2");
+ JobVertex v1 = new JobVertex("vertex1");
+ JobVertex v2 = new JobVertex("vertex2");
v1.setParallelism(N);
v2.setParallelism(7 * N);
v2.connectNewDataSetAsInput(v1, DistributionPattern.POINTWISE);
- List<AbstractJobVertex> ordered = new ArrayList<AbstractJobVertex>(Arrays.asList(v1, v2));
+ List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2));
ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
try {
@@ -244,15 +244,15 @@ public class PointwisePatternTest {
final int factor = highDop / lowDop;
final int delta = highDop % lowDop == 0 ? 0 : 1;
- AbstractJobVertex v1 = new AbstractJobVertex("vertex1");
- AbstractJobVertex v2 = new AbstractJobVertex("vertex2");
+ JobVertex v1 = new JobVertex("vertex1");
+ JobVertex v2 = new JobVertex("vertex2");
v1.setParallelism(lowDop);
v2.setParallelism(highDop);
v2.connectNewDataSetAsInput(v1, DistributionPattern.POINTWISE);
- List<AbstractJobVertex> ordered = new ArrayList<AbstractJobVertex>(Arrays.asList(v1, v2));
+ List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2));
ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
try {
@@ -290,15 +290,15 @@ public class PointwisePatternTest {
final int factor = highDop / lowDop;
final int delta = highDop % lowDop == 0 ? 0 : 1;
- AbstractJobVertex v1 = new AbstractJobVertex("vertex1");
- AbstractJobVertex v2 = new AbstractJobVertex("vertex2");
+ JobVertex v1 = new JobVertex("vertex1");
+ JobVertex v2 = new JobVertex("vertex2");
v1.setParallelism(highDop);
v2.setParallelism(lowDop);
v2.connectNewDataSetAsInput(v1, DistributionPattern.POINTWISE);
- List<AbstractJobVertex> ordered = new ArrayList<AbstractJobVertex>(Arrays.asList(v1, v2));
+ List<JobVertex> ordered = new ArrayList<JobVertex>(Arrays.asList(v1, v2));
ExecutionGraph eg = new ExecutionGraph(jobId, jobName, cfg, AkkaUtils.getDefaultTimeout());
try {
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TerminalStateDeadlockTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TerminalStateDeadlockTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TerminalStateDeadlockTest.java
index 8cba6ca..376ff14 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TerminalStateDeadlockTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TerminalStateDeadlockTest.java
@@ -28,7 +28,7 @@ import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.instance.InstanceConnectionInfo;
import org.apache.flink.runtime.instance.InstanceID;
import org.apache.flink.runtime.instance.SimpleSlot;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobmanager.scheduler.Scheduler;
@@ -105,10 +105,10 @@ public class TerminalStateDeadlockTest {
final Configuration jobConfig = new Configuration();
- final List<AbstractJobVertex> vertices;
+ final List<JobVertex> vertices;
{
- AbstractJobVertex v1 = new AbstractJobVertex("v1", vid1);
- AbstractJobVertex v2 = new AbstractJobVertex("v2", vid2);
+ JobVertex v1 = new JobVertex("v1", vid1);
+ JobVertex v2 = new JobVertex("v2", vid2);
v1.setParallelism(1);
v2.setParallelism(1);
v1.setInvokableClass(DummyInvokable.class);
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexLocationConstraintTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexLocationConstraintTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexLocationConstraintTest.java
index cae10f4..756b9a4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexLocationConstraintTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexLocationConstraintTest.java
@@ -31,7 +31,7 @@ import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.instance.InstanceConnectionInfo;
import org.apache.flink.runtime.instance.InstanceID;
import org.apache.flink.runtime.instance.SimpleSlot;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException;
@@ -97,7 +97,7 @@ public class VertexLocationConstraintTest {
scheduler.newInstanceAvailable(instance3);
// prepare the execution graph
- AbstractJobVertex jobVertex = new AbstractJobVertex("test vertex", new JobVertexID());
+ JobVertex jobVertex = new JobVertex("test vertex", new JobVertexID());
jobVertex.setInvokableClass(DummyInvokable.class);
jobVertex.setParallelism(2);
JobGraph jg = new JobGraph("test job", jobVertex);
@@ -163,7 +163,7 @@ public class VertexLocationConstraintTest {
scheduler.newInstanceAvailable(instance3);
// prepare the execution graph
- AbstractJobVertex jobVertex = new AbstractJobVertex("test vertex", new JobVertexID());
+ JobVertex jobVertex = new JobVertex("test vertex", new JobVertexID());
jobVertex.setInvokableClass(DummyInvokable.class);
jobVertex.setParallelism(2);
JobGraph jg = new JobGraph("test job", jobVertex);
@@ -225,8 +225,8 @@ public class VertexLocationConstraintTest {
scheduler.newInstanceAvailable(instance3);
// prepare the execution graph
- AbstractJobVertex jobVertex1 = new AbstractJobVertex("v1", new JobVertexID());
- AbstractJobVertex jobVertex2 = new AbstractJobVertex("v2", new JobVertexID());
+ JobVertex jobVertex1 = new JobVertex("v1", new JobVertexID());
+ JobVertex jobVertex2 = new JobVertex("v2", new JobVertexID());
jobVertex1.setInvokableClass(DummyInvokable.class);
jobVertex2.setInvokableClass(DummyInvokable.class);
jobVertex1.setParallelism(2);
@@ -294,7 +294,7 @@ public class VertexLocationConstraintTest {
scheduler.newInstanceAvailable(instance1);
// prepare the execution graph
- AbstractJobVertex jobVertex = new AbstractJobVertex("test vertex", new JobVertexID());
+ JobVertex jobVertex = new JobVertex("test vertex", new JobVertexID());
jobVertex.setInvokableClass(DummyInvokable.class);
jobVertex.setParallelism(1);
JobGraph jg = new JobGraph("test job", jobVertex);
@@ -347,8 +347,8 @@ public class VertexLocationConstraintTest {
scheduler.newInstanceAvailable(instance1);
// prepare the execution graph
- AbstractJobVertex jobVertex1 = new AbstractJobVertex("v1", new JobVertexID());
- AbstractJobVertex jobVertex2 = new AbstractJobVertex("v2", new JobVertexID());
+ JobVertex jobVertex1 = new JobVertex("v1", new JobVertexID());
+ JobVertex jobVertex2 = new JobVertex("v2", new JobVertexID());
jobVertex1.setInvokableClass(DummyInvokable.class);
jobVertex2.setInvokableClass(DummyInvokable.class);
@@ -392,7 +392,7 @@ public class VertexLocationConstraintTest {
@Test
public void testArchivingClearsFields() {
try {
- AbstractJobVertex vertex = new AbstractJobVertex("test vertex", new JobVertexID());
+ JobVertex vertex = new JobVertex("test vertex", new JobVertexID());
JobGraph jg = new JobGraph("test job", vertex);
ExecutionGraph eg = new ExecutionGraph(jg.getJobID(), jg.getName(), jg.getJobConfiguration(), timeout);
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexSlotSharingTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexSlotSharingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexSlotSharingTest.java
index 17d78d2..a1d6d03 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexSlotSharingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexSlotSharingTest.java
@@ -26,7 +26,7 @@ import java.util.List;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.akka.AkkaUtils;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
@@ -43,11 +43,11 @@ public class VertexSlotSharingTest {
@Test
public void testAssignSlotSharingGroup() {
try {
- AbstractJobVertex v1 = new AbstractJobVertex("v1");
- AbstractJobVertex v2 = new AbstractJobVertex("v2");
- AbstractJobVertex v3 = new AbstractJobVertex("v3");
- AbstractJobVertex v4 = new AbstractJobVertex("v4");
- AbstractJobVertex v5 = new AbstractJobVertex("v5");
+ JobVertex v1 = new JobVertex("v1");
+ JobVertex v2 = new JobVertex("v2");
+ JobVertex v3 = new JobVertex("v3");
+ JobVertex v4 = new JobVertex("v4");
+ JobVertex v5 = new JobVertex("v5");
v1.setParallelism(4);
v2.setParallelism(5);
@@ -66,7 +66,7 @@ public class VertexSlotSharingTest {
v4.setSlotSharingGroup(jg2);
v5.setSlotSharingGroup(jg2);
- List<AbstractJobVertex> vertices = new ArrayList<AbstractJobVertex>(Arrays.asList(v1, v2, v3, v4, v5));
+ List<JobVertex> vertices = new ArrayList<JobVertex>(Arrays.asList(v1, v2, v3, v4, v5));
ExecutionGraph eg = new ExecutionGraph(new JobID(), "test job", new Configuration(),
AkkaUtils.getDefaultTimeout());
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SharedSlotsTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SharedSlotsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SharedSlotsTest.java
index e49f19f..8709395 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SharedSlotsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/instance/SharedSlotsTest.java
@@ -19,7 +19,7 @@
package org.apache.flink.runtime.instance;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobmanager.scheduler.CoLocationConstraint;
import org.apache.flink.runtime.jobmanager.scheduler.CoLocationGroup;
@@ -409,8 +409,8 @@ public class SharedSlotsTest {
JobVertexID tailId = new JobVertexID();
JobVertexID sinkId = new JobVertexID();
- AbstractJobVertex headVertex = new AbstractJobVertex("head", headId);
- AbstractJobVertex tailVertex = new AbstractJobVertex("tail", tailId);
+ JobVertex headVertex = new JobVertex("head", headId);
+ JobVertex tailVertex = new JobVertex("tail", tailId);
SlotSharingGroup sharingGroup = new SlotSharingGroup(sourceId, headId, tailId, sinkId);
SlotSharingGroupAssignment assignment = sharingGroup.getTaskAssignment();
@@ -545,8 +545,8 @@ public class SharedSlotsTest {
JobVertexID tailId = new JobVertexID();
JobVertexID sinkId = new JobVertexID();
- AbstractJobVertex headVertex = new AbstractJobVertex("head", headId);
- AbstractJobVertex tailVertex = new AbstractJobVertex("tail", tailId);
+ JobVertex headVertex = new JobVertex("head", headId);
+ JobVertex tailVertex = new JobVertex("tail", tailId);
SlotSharingGroup sharingGroup = new SlotSharingGroup(sourceId, headId, tailId, sinkId);
SlotSharingGroupAssignment assignment = sharingGroup.getTaskAssignment();
@@ -633,7 +633,7 @@ public class SharedSlotsTest {
try {
JobID jobId = new JobID();
JobVertexID vid = new JobVertexID();
- AbstractJobVertex vertex = new AbstractJobVertex("vertex", vid);
+ JobVertex vertex = new JobVertex("vertex", vid);
SlotSharingGroup sharingGroup = new SlotSharingGroup(vid);
SlotSharingGroupAssignment assignment = sharingGroup.getTaskAssignment();
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartialConsumePipelinedResultTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartialConsumePipelinedResultTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartialConsumePipelinedResultTest.java
index 821826a..7e15af4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartialConsumePipelinedResultTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartialConsumePipelinedResultTest.java
@@ -25,7 +25,7 @@ import org.apache.flink.runtime.client.JobClient;
import org.apache.flink.runtime.io.network.api.reader.BufferReader;
import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
import org.apache.flink.runtime.io.network.buffer.Buffer;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
@@ -78,11 +78,11 @@ public class PartialConsumePipelinedResultTest {
*/
@Test
public void testPartialConsumePipelinedResultReceiver() throws Exception {
- final AbstractJobVertex sender = new AbstractJobVertex("Sender");
+ final JobVertex sender = new JobVertex("Sender");
sender.setInvokableClass(SlowBufferSender.class);
sender.setParallelism(PARALLELISM);
- final AbstractJobVertex receiver = new AbstractJobVertex("Receiver");
+ final JobVertex receiver = new JobVertex("Receiver");
receiver.setInvokableClass(SingleBufferReceiver.class);
receiver.setParallelism(PARALLELISM);
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobGraphTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobGraphTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobGraphTest.java
index aac6948..9f88bd5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobGraphTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobGraphTest.java
@@ -41,9 +41,9 @@ public class JobGraphTest {
// add some vertices
{
- AbstractJobVertex source1 = new AbstractJobVertex("source1");
- AbstractJobVertex source2 = new AbstractJobVertex("source2");
- AbstractJobVertex target = new AbstractJobVertex("target");
+ JobVertex source1 = new JobVertex("source1");
+ JobVertex source2 = new JobVertex("source2");
+ JobVertex target = new JobVertex("target");
target.connectNewDataSetAsInput(source1, DistributionPattern.POINTWISE);
target.connectNewDataSetAsInput(source2, DistributionPattern.ALL_TO_ALL);
@@ -60,8 +60,8 @@ public class JobGraphTest {
assertEquals(jg.getJobConfiguration(), copy.getJobConfiguration());
assertEquals(jg.getNumberOfVertices(), copy.getNumberOfVertices());
- for (AbstractJobVertex vertex : copy.getVertices()) {
- AbstractJobVertex original = jg.findVertexByID(vertex.getID());
+ for (JobVertex vertex : copy.getVertices()) {
+ JobVertex original = jg.findVertexByID(vertex.getID());
assertNotNull(original);
assertEquals(original.getName(), vertex.getName());
assertEquals(original.getNumberOfInputs(), vertex.getNumberOfInputs());
@@ -77,12 +77,12 @@ public class JobGraphTest {
@Test
public void testTopologicalSort1() {
try {
- AbstractJobVertex source1 = new AbstractJobVertex("source1");
- AbstractJobVertex source2 = new AbstractJobVertex("source2");
- AbstractJobVertex target1 = new AbstractJobVertex("target1");
- AbstractJobVertex target2 = new AbstractJobVertex("target2");
- AbstractJobVertex intermediate1 = new AbstractJobVertex("intermediate1");
- AbstractJobVertex intermediate2 = new AbstractJobVertex("intermediate2");
+ JobVertex source1 = new JobVertex("source1");
+ JobVertex source2 = new JobVertex("source2");
+ JobVertex target1 = new JobVertex("target1");
+ JobVertex target2 = new JobVertex("target2");
+ JobVertex intermediate1 = new JobVertex("intermediate1");
+ JobVertex intermediate2 = new JobVertex("intermediate2");
target1.connectNewDataSetAsInput(source1, DistributionPattern.POINTWISE);
target2.connectNewDataSetAsInput(source1, DistributionPattern.POINTWISE);
@@ -91,7 +91,7 @@ public class JobGraphTest {
intermediate1.connectNewDataSetAsInput(source2, DistributionPattern.POINTWISE);
JobGraph graph = new JobGraph("TestGraph", source1, source2, intermediate1, intermediate2, target1, target2);
- List<AbstractJobVertex> sorted = graph.getVerticesSortedTopologicallyFromSources();
+ List<JobVertex> sorted = graph.getVerticesSortedTopologicallyFromSources();
assertEquals(6, sorted.size());
@@ -112,13 +112,13 @@ public class JobGraphTest {
@Test
public void testTopologicalSort2() {
try {
- AbstractJobVertex source1 = new AbstractJobVertex("source1");
- AbstractJobVertex source2 = new AbstractJobVertex("source2");
- AbstractJobVertex root = new AbstractJobVertex("root");
- AbstractJobVertex l11 = new AbstractJobVertex("layer 1 - 1");
- AbstractJobVertex l12 = new AbstractJobVertex("layer 1 - 2");
- AbstractJobVertex l13 = new AbstractJobVertex("layer 1 - 3");
- AbstractJobVertex l2 = new AbstractJobVertex("layer 2");
+ JobVertex source1 = new JobVertex("source1");
+ JobVertex source2 = new JobVertex("source2");
+ JobVertex root = new JobVertex("root");
+ JobVertex l11 = new JobVertex("layer 1 - 1");
+ JobVertex l12 = new JobVertex("layer 1 - 2");
+ JobVertex l13 = new JobVertex("layer 1 - 3");
+ JobVertex l2 = new JobVertex("layer 2");
root.connectNewDataSetAsInput(l13, DistributionPattern.POINTWISE);
root.connectNewDataSetAsInput(source2, DistributionPattern.POINTWISE);
@@ -135,7 +135,7 @@ public class JobGraphTest {
l13.connectNewDataSetAsInput(source2, DistributionPattern.POINTWISE);
JobGraph graph = new JobGraph("TestGraph", source1, source2, root, l11, l13, l12, l2);
- List<AbstractJobVertex> sorted = graph.getVerticesSortedTopologicallyFromSources();
+ List<JobVertex> sorted = graph.getVerticesSortedTopologicallyFromSources();
assertEquals(7, sorted.size());
@@ -170,10 +170,10 @@ public class JobGraphTest {
// ---------
try {
- AbstractJobVertex source = new AbstractJobVertex("source");
- AbstractJobVertex op1 = new AbstractJobVertex("op4");
- AbstractJobVertex op2 = new AbstractJobVertex("op2");
- AbstractJobVertex op3 = new AbstractJobVertex("op3");
+ JobVertex source = new JobVertex("source");
+ JobVertex op1 = new JobVertex("op4");
+ JobVertex op2 = new JobVertex("op2");
+ JobVertex op3 = new JobVertex("op3");
op1.connectNewDataSetAsInput(source, DistributionPattern.POINTWISE);
op2.connectNewDataSetAsInput(op1, DistributionPattern.POINTWISE);
@@ -181,7 +181,7 @@ public class JobGraphTest {
op3.connectNewDataSetAsInput(op2, DistributionPattern.POINTWISE);
JobGraph graph = new JobGraph("TestGraph", source, op1, op2, op3);
- List<AbstractJobVertex> sorted = graph.getVerticesSortedTopologicallyFromSources();
+ List<JobVertex> sorted = graph.getVerticesSortedTopologicallyFromSources();
assertEquals(4, sorted.size());
@@ -199,10 +199,10 @@ public class JobGraphTest {
@Test
public void testTopoSortCyclicGraphNoSources() {
try {
- AbstractJobVertex v1 = new AbstractJobVertex("1");
- AbstractJobVertex v2 = new AbstractJobVertex("2");
- AbstractJobVertex v3 = new AbstractJobVertex("3");
- AbstractJobVertex v4 = new AbstractJobVertex("4");
+ JobVertex v1 = new JobVertex("1");
+ JobVertex v2 = new JobVertex("2");
+ JobVertex v3 = new JobVertex("3");
+ JobVertex v4 = new JobVertex("4");
v1.connectNewDataSetAsInput(v4, DistributionPattern.POINTWISE);
v2.connectNewDataSetAsInput(v1, DistributionPattern.POINTWISE);
@@ -227,12 +227,12 @@ public class JobGraphTest {
@Test
public void testTopoSortCyclicGraphIntermediateCycle() {
try{
- AbstractJobVertex source = new AbstractJobVertex("source");
- AbstractJobVertex v1 = new AbstractJobVertex("1");
- AbstractJobVertex v2 = new AbstractJobVertex("2");
- AbstractJobVertex v3 = new AbstractJobVertex("3");
- AbstractJobVertex v4 = new AbstractJobVertex("4");
- AbstractJobVertex target = new AbstractJobVertex("target");
+ JobVertex source = new JobVertex("source");
+ JobVertex v1 = new JobVertex("1");
+ JobVertex v2 = new JobVertex("2");
+ JobVertex v3 = new JobVertex("3");
+ JobVertex v4 = new JobVertex("4");
+ JobVertex target = new JobVertex("target");
v1.connectNewDataSetAsInput(source, DistributionPattern.POINTWISE);
v1.connectNewDataSetAsInput(v4, DistributionPattern.POINTWISE);
@@ -256,9 +256,9 @@ public class JobGraphTest {
}
}
- private static final void assertBefore(AbstractJobVertex v1, AbstractJobVertex v2, List<AbstractJobVertex> list) {
+ private static final void assertBefore(JobVertex v1, JobVertex v2, List<JobVertex> list) {
boolean seenFirst = false;
- for (AbstractJobVertex v : list) {
+ for (JobVertex v : list) {
if (v == v1) {
seenFirst = true;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobTaskVertexTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobTaskVertexTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobTaskVertexTest.java
index b1d05e1..2511bd6 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobTaskVertexTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobTaskVertexTest.java
@@ -40,8 +40,8 @@ public class JobTaskVertexTest {
@Test
public void testConnectDirectly() {
- AbstractJobVertex source = new AbstractJobVertex("source");
- AbstractJobVertex target = new AbstractJobVertex("target");
+ JobVertex source = new JobVertex("source");
+ JobVertex target = new JobVertex("target");
target.connectNewDataSetAsInput(source, DistributionPattern.POINTWISE);
assertTrue(source.isInputVertex());
@@ -60,9 +60,9 @@ public class JobTaskVertexTest {
@Test
public void testConnectMultipleTargets() {
- AbstractJobVertex source = new AbstractJobVertex("source");
- AbstractJobVertex target1= new AbstractJobVertex("target1");
- AbstractJobVertex target2 = new AbstractJobVertex("target2");
+ JobVertex source = new JobVertex("source");
+ JobVertex target1= new JobVertex("target1");
+ JobVertex target2 = new JobVertex("target2");
target1.connectNewDataSetAsInput(source, DistributionPattern.POINTWISE);
target2.connectDataSetAsInput(source.getProducedDataSets().get(0), DistributionPattern.ALL_TO_ALL);
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerTest.java
index 9dd078d..3eb4be8 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerTest.java
@@ -33,7 +33,7 @@ import org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils;
import org.apache.flink.runtime.executiongraph.ExecutionVertex;
import org.apache.flink.runtime.executiongraph.IntermediateResultPartition;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.messages.JobManagerMessages;
@@ -105,7 +105,7 @@ public class JobManagerTest {
final IntermediateDataSetID rid = new IntermediateDataSetID();
// Create a task
- final AbstractJobVertex sender = new AbstractJobVertex("Sender");
+ final JobVertex sender = new JobVertex("Sender");
sender.setParallelism(1);
sender.setInvokableClass(Tasks.BlockingNoOpInvokable.class); // just block
sender.createAndAddResultDataSet(rid, PIPELINED);
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobSubmitTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobSubmitTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobSubmitTest.java
index 7b6d688..2f11c08 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobSubmitTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobSubmitTest.java
@@ -27,7 +27,7 @@ import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.blob.BlobClient;
import org.apache.flink.runtime.blob.BlobKey;
import org.apache.flink.runtime.client.JobExecutionException;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.messages.JobManagerMessages;
import org.junit.AfterClass;
@@ -76,7 +76,7 @@ public class JobSubmitTest {
public void testFailureWhenJarBlobsMissing() {
try {
// create a simple job graph
- AbstractJobVertex jobVertex = new AbstractJobVertex("Test Vertex");
+ JobVertex jobVertex = new JobVertex("Test Vertex");
jobVertex.setInvokableClass(Tasks.NoOpInvokable.class);
JobGraph jg = new JobGraph("test job", jobVertex);
@@ -130,7 +130,7 @@ public class JobSubmitTest {
try {
// create a simple job graph
- AbstractJobVertex jobVertex = new AbstractJobVertex("Vertex that fails in initializeOnMaster") {
+ JobVertex jobVertex = new JobVertex("Vertex that fails in initializeOnMaster") {
@Override
public void initializeOnMaster(ClassLoader loader) throws Exception {
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/SlotCountExceedingParallelismTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/SlotCountExceedingParallelismTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/SlotCountExceedingParallelismTest.java
index b404aae..8272bf7 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/SlotCountExceedingParallelismTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/SlotCountExceedingParallelismTest.java
@@ -22,7 +22,7 @@ import org.apache.flink.runtime.client.JobExecutionException;
import org.apache.flink.runtime.io.network.api.reader.RecordReader;
import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
@@ -86,12 +86,12 @@ public class SlotCountExceedingParallelismTest {
int receiverParallelism) {
// The sender and receiver invokable logic ensure that each subtask gets the expected data
- final AbstractJobVertex sender = new AbstractJobVertex("Sender");
+ final JobVertex sender = new JobVertex("Sender");
sender.setInvokableClass(RoundRobinSubtaskIndexSender.class);
sender.getConfiguration().setInteger(RoundRobinSubtaskIndexSender.CONFIG_KEY, receiverParallelism);
sender.setParallelism(senderParallelism);
- final AbstractJobVertex receiver = new AbstractJobVertex("Receiver");
+ final JobVertex receiver = new JobVertex("Receiver");
receiver.setInvokableClass(SubtaskIndexReceiver.class);
receiver.getConfiguration().setInteger(SubtaskIndexReceiver.CONFIG_KEY, senderParallelism);
receiver.setParallelism(receiverParallelism);
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationConstraintTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationConstraintTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationConstraintTest.java
index 94dee71..3bd4368 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationConstraintTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationConstraintTest.java
@@ -22,7 +22,7 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.instance.SharedSlot;
import org.apache.flink.runtime.instance.SlotSharingGroupAssignment;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.util.AbstractID;
import org.junit.Test;
@@ -40,10 +40,10 @@ public class CoLocationConstraintTest {
JobVertexID id1 = new JobVertexID();
JobVertexID id2 = new JobVertexID();
- AbstractJobVertex vertex1 = new AbstractJobVertex("vertex1", id1);
+ JobVertex vertex1 = new JobVertex("vertex1", id1);
vertex1.setParallelism(2);
- AbstractJobVertex vertex2 = new AbstractJobVertex("vertex2", id2);
+ JobVertex vertex2 = new JobVertex("vertex2", id2);
vertex2.setParallelism(3);
CoLocationGroup group = new CoLocationGroup(vertex1, vertex2);
@@ -74,7 +74,7 @@ public class CoLocationConstraintTest {
try {
JobID jid = new JobID();
- AbstractJobVertex vertex = new AbstractJobVertex("vertex");
+ JobVertex vertex = new JobVertex("vertex");
vertex.setParallelism(1);
SlotSharingGroup sharingGroup = new SlotSharingGroup(vertex.getID());
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/ScheduleOrUpdateConsumersTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/ScheduleOrUpdateConsumersTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/ScheduleOrUpdateConsumersTest.java
index 08f8bfb..3bcac56 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/ScheduleOrUpdateConsumersTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/ScheduleOrUpdateConsumersTest.java
@@ -22,7 +22,7 @@ import com.google.common.collect.Lists;
import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
@@ -79,12 +79,12 @@ public class ScheduleOrUpdateConsumersTest {
*/
@Test
public void testMixedPipelinedAndBlockingResults() throws Exception {
- final AbstractJobVertex sender = new AbstractJobVertex("Sender");
+ final JobVertex sender = new JobVertex("Sender");
sender.setInvokableClass(BinaryRoundRobinSubtaskIndexSender.class);
sender.getConfiguration().setInteger(BinaryRoundRobinSubtaskIndexSender.CONFIG_KEY, PARALLELISM);
sender.setParallelism(PARALLELISM);
- final AbstractJobVertex pipelinedReceiver = new AbstractJobVertex("Pipelined Receiver");
+ final JobVertex pipelinedReceiver = new JobVertex("Pipelined Receiver");
pipelinedReceiver.setInvokableClass(SlotCountExceedingParallelismTest.SubtaskIndexReceiver.class);
pipelinedReceiver.getConfiguration().setInteger(CONFIG_KEY, PARALLELISM);
pipelinedReceiver.setParallelism(PARALLELISM);
@@ -94,7 +94,7 @@ public class ScheduleOrUpdateConsumersTest {
DistributionPattern.ALL_TO_ALL,
ResultPartitionType.PIPELINED);
- final AbstractJobVertex blockingReceiver = new AbstractJobVertex("Blocking Receiver");
+ final JobVertex blockingReceiver = new JobVertex("Blocking Receiver");
blockingReceiver.setInvokableClass(SlotCountExceedingParallelismTest.SubtaskIndexReceiver.class);
blockingReceiver.getConfiguration().setInteger(CONFIG_KEY, PARALLELISM);
blockingReceiver.setParallelism(PARALLELISM);
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelTest.java
index 77193f1..67a5f44 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskCancelTest.java
@@ -29,7 +29,7 @@ import org.apache.flink.runtime.io.network.api.reader.RecordReader;
import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
import org.apache.flink.runtime.io.network.partition.consumer.UnionInputGate;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobStatus;
@@ -77,12 +77,12 @@ public class TaskCancelTest {
// Setup
final JobGraph jobGraph = new JobGraph("Cancel Big Union");
- AbstractJobVertex[] sources = new AbstractJobVertex[numberOfSources];
+ JobVertex[] sources = new JobVertex[numberOfSources];
SlotSharingGroup group = new SlotSharingGroup();
// Create multiple sources
for (int i = 0; i < sources.length; i++) {
- sources[i] = new AbstractJobVertex("Source " + i);
+ sources[i] = new JobVertex("Source " + i);
sources[i].setInvokableClass(InfiniteSource.class);
sources[i].setParallelism(sourceParallelism);
sources[i].setSlotSharingGroup(group);
@@ -92,14 +92,14 @@ public class TaskCancelTest {
}
// Union all sources
- AbstractJobVertex union = new AbstractJobVertex("Union");
+ JobVertex union = new JobVertex("Union");
union.setInvokableClass(AgnosticUnion.class);
union.setParallelism(sourceParallelism);
jobGraph.addVertex(union);
// Each source creates a separate result
- for (AbstractJobVertex source : sources) {
+ for (JobVertex source : sources) {
union.connectNewDataSetAsInput(
source,
DistributionPattern.POINTWISE,
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.scala
index a36ded9..a1ca43c 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.scala
@@ -23,7 +23,7 @@ import akka.testkit.TestKit
import org.apache.flink.api.common.JobID
import org.apache.flink.configuration.Configuration
import org.apache.flink.runtime.akka.AkkaUtils
-import org.apache.flink.runtime.jobgraph.{JobStatus, JobGraph, AbstractJobVertex}
+import org.apache.flink.runtime.jobgraph.{JobStatus, JobGraph, JobVertex}
import org.apache.flink.runtime.jobmanager.Tasks
import org.apache.flink.runtime.jobmanager.scheduler.Scheduler
import org.apache.flink.runtime.testingUtils.TestingUtils
@@ -50,7 +50,7 @@ with Matchers with BeforeAndAfterAll {
val scheduler = new Scheduler
scheduler.newInstanceAvailable(instance)
- val sender = new AbstractJobVertex("Task")
+ val sender = new JobVertex("Task")
sender.setInvokableClass(classOf[Tasks.NoOpInvokable])
sender.setParallelism(NUM_TASKS)
@@ -94,7 +94,7 @@ with Matchers with BeforeAndAfterAll {
val scheduler = new Scheduler
scheduler.newInstanceAvailable(instance)
- val sender = new AbstractJobVertex("Task")
+ val sender = new JobVertex("Task")
sender.setInvokableClass(classOf[Tasks.NoOpInvokable])
sender.setParallelism(NUM_TASKS)
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/TaskManagerLossFailsTasksTest.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/TaskManagerLossFailsTasksTest.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/TaskManagerLossFailsTasksTest.scala
index 1a36112..13199bc 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/TaskManagerLossFailsTasksTest.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/TaskManagerLossFailsTasksTest.scala
@@ -25,7 +25,7 @@ import org.apache.flink.configuration.Configuration
import org.apache.flink.runtime.akka.AkkaUtils
import org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils
.SimpleAcknowledgingTaskManager
-import org.apache.flink.runtime.jobgraph.{JobStatus, JobGraph, AbstractJobVertex}
+import org.apache.flink.runtime.jobgraph.{JobStatus, JobGraph, JobVertex}
import org.apache.flink.runtime.jobmanager.Tasks
import org.apache.flink.runtime.jobmanager.scheduler.Scheduler
import org.apache.flink.runtime.testingUtils.TestingUtils
@@ -53,7 +53,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
scheduler.newInstanceAvailable(instance1)
scheduler.newInstanceAvailable(instance2)
- val sender = new AbstractJobVertex("Task")
+ val sender = new JobVertex("Task")
sender.setInvokableClass(classOf[Tasks.NoOpInvokable])
sender.setParallelism(20)
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/CoLocationConstraintITCase.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/CoLocationConstraintITCase.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/CoLocationConstraintITCase.scala
index 07f0ce5..19f1ea1 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/CoLocationConstraintITCase.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/CoLocationConstraintITCase.scala
@@ -21,8 +21,7 @@ package org.apache.flink.runtime.jobmanager
import akka.actor.ActorSystem
import akka.actor.Status.Success
import akka.testkit.{ImplicitSender, TestKit}
-import org.apache.flink.runtime.jobgraph.{JobGraph, DistributionPattern,
-AbstractJobVertex}
+import org.apache.flink.runtime.jobgraph.{JobGraph, DistributionPattern, JobVertex}
import org.apache.flink.runtime.jobmanager.Tasks.{Receiver, Sender}
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup
import org.apache.flink.runtime.messages.JobManagerMessages.{JobResultSuccess, SubmitJob}
@@ -45,8 +44,8 @@ ImplicitSender with WordSpecLike with Matchers with BeforeAndAfterAll with WrapA
"support colocation constraints and slot sharing" in {
val num_tasks = 31
- val sender = new AbstractJobVertex("Sender")
- val receiver = new AbstractJobVertex("Receiver")
+ val sender = new JobVertex("Sender")
+ val receiver = new JobVertex("Receiver")
sender.setInvokableClass(classOf[Sender])
receiver.setInvokableClass(classOf[Receiver])
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala
index ee584f0..750da83 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerITCase.scala
@@ -25,7 +25,7 @@ import akka.pattern.ask
import akka.testkit.{ImplicitSender, TestKit}
import akka.util.Timeout
import org.apache.flink.runtime.client.JobExecutionException
-import org.apache.flink.runtime.jobgraph.{AbstractJobVertex, DistributionPattern, JobGraph, ScheduleMode}
+import org.apache.flink.runtime.jobgraph.{JobVertex, DistributionPattern, JobGraph, ScheduleMode}
import org.apache.flink.runtime.messages.JobManagerMessages._
import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages.NotifyWhenJobRemoved
import org.apache.flink.runtime.testingUtils.TestingUtils
@@ -54,7 +54,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
"The JobManager actor" must {
"handle jobs when not enough slots" in {
- val vertex = new AbstractJobVertex("Test Vertex")
+ val vertex = new JobVertex("Test Vertex")
vertex.setParallelism(2)
vertex.setInvokableClass(classOf[BlockingNoOpInvokable])
@@ -99,7 +99,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
"support immediate scheduling of a single vertex" in {
val num_tasks = 133
- val vertex = new AbstractJobVertex("Test Vertex")
+ val vertex = new JobVertex("Test Vertex")
vertex.setParallelism(num_tasks)
vertex.setInvokableClass(classOf[NoOpInvokable])
@@ -134,7 +134,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
"support queued scheduling of a single vertex" in {
val num_tasks = 111
- val vertex = new AbstractJobVertex("Test Vertex")
+ val vertex = new JobVertex("Test Vertex")
vertex.setParallelism(num_tasks)
vertex.setInvokableClass(classOf[NoOpInvokable])
@@ -163,8 +163,8 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
"support forward jobs" in {
val num_tasks = 31
- val sender = new AbstractJobVertex("Sender")
- val receiver = new AbstractJobVertex("Receiver")
+ val sender = new JobVertex("Sender")
+ val receiver = new JobVertex("Receiver")
sender.setInvokableClass(classOf[Sender])
receiver.setInvokableClass(classOf[Receiver])
@@ -198,8 +198,8 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
"support bipartite job" in {
val num_tasks = 31
- val sender = new AbstractJobVertex("Sender")
- val receiver = new AbstractJobVertex("Receiver")
+ val sender = new JobVertex("Sender")
+ val receiver = new JobVertex("Receiver")
sender.setInvokableClass(classOf[Sender])
receiver.setInvokableClass(classOf[AgnosticReceiver])
@@ -231,9 +231,9 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
"support two input job failing edge mismatch" in {
val num_tasks = 1
- val sender1 = new AbstractJobVertex("Sender1")
- val sender2 = new AbstractJobVertex("Sender2")
- val receiver = new AbstractJobVertex("Receiver")
+ val sender1 = new JobVertex("Sender1")
+ val sender2 = new JobVertex("Sender2")
+ val receiver = new JobVertex("Receiver")
sender1.setInvokableClass(classOf[Sender])
sender2.setInvokableClass(classOf[Sender])
@@ -275,9 +275,9 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
"support two input job" in {
val num_tasks = 11
- val sender1 = new AbstractJobVertex("Sender1")
- val sender2 = new AbstractJobVertex("Sender2")
- val receiver = new AbstractJobVertex("Receiver")
+ val sender1 = new JobVertex("Sender1")
+ val sender2 = new JobVertex("Sender2")
+ val receiver = new JobVertex("Receiver")
sender1.setInvokableClass(classOf[Sender])
sender2.setInvokableClass(classOf[Sender])
@@ -312,9 +312,9 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
"support scheduling all at once" in {
val num_tasks = 16
- val sender = new AbstractJobVertex("Sender")
- val forwarder = new AbstractJobVertex("Forwarder")
- val receiver = new AbstractJobVertex("Receiver")
+ val sender = new JobVertex("Sender")
+ val forwarder = new JobVertex("Forwarder")
+ val receiver = new JobVertex("Receiver")
sender.setInvokableClass(classOf[Sender])
forwarder.setInvokableClass(classOf[Forwarder])
@@ -357,8 +357,8 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
"handle job with a failing sender vertex" in {
val num_tasks = 100
- val sender = new AbstractJobVertex("Sender")
- val receiver = new AbstractJobVertex("Receiver")
+ val sender = new JobVertex("Sender")
+ val receiver = new JobVertex("Receiver")
sender.setInvokableClass(classOf[ExceptionSender])
receiver.setInvokableClass(classOf[Receiver])
@@ -402,8 +402,8 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
"handle job with an occasionally failing sender vertex" in {
val num_tasks = 100
- val sender = new AbstractJobVertex("Sender")
- val receiver = new AbstractJobVertex("Receiver")
+ val sender = new JobVertex("Sender")
+ val receiver = new JobVertex("Receiver")
sender.setInvokableClass(classOf[SometimesExceptionSender])
receiver.setInvokableClass(classOf[Receiver])
@@ -449,8 +449,8 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
"handle job with a failing receiver vertex" in {
val num_tasks = 200
- val sender = new AbstractJobVertex("Sender")
- val receiver = new AbstractJobVertex("Receiver")
+ val sender = new JobVertex("Sender")
+ val receiver = new JobVertex("Receiver")
sender.setInvokableClass(classOf[Sender])
receiver.setInvokableClass(classOf[ExceptionReceiver])
@@ -488,8 +488,8 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
"handle job with all vertices failing during instantiation" in {
val num_tasks = 200
- val sender = new AbstractJobVertex("Sender")
- val receiver = new AbstractJobVertex("Receiver")
+ val sender = new JobVertex("Sender")
+ val receiver = new JobVertex("Receiver")
sender.setInvokableClass(classOf[InstantiationErrorSender])
receiver.setInvokableClass(classOf[Receiver])
@@ -531,8 +531,8 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
"handle job with some vertices failing during instantiation" in {
val num_tasks = 200
- val sender = new AbstractJobVertex("Sender")
- val receiver = new AbstractJobVertex("Receiver")
+ val sender = new JobVertex("Sender")
+ val receiver = new JobVertex("Receiver")
sender.setInvokableClass(classOf[SometimesInstantiationErrorSender])
receiver.setInvokableClass(classOf[Receiver])
@@ -579,7 +579,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
"completes" in {
val num_tasks = 31
- val source = new AbstractJobVertex("Source")
+ val source = new JobVertex("Source")
val sink = new WaitingOnFinalizeJobVertex("Sink", 500)
source.setInvokableClass(classOf[WaitingNoOpInvokable])
@@ -611,8 +611,7 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
}
}
- class WaitingOnFinalizeJobVertex(name: String, val waitingTime: Long) extends
- AbstractJobVertex(name){
+ class WaitingOnFinalizeJobVertex(name: String, val waitingTime: Long) extends JobVertex(name){
var finished = false
override def finalizeOnMaster(loader: ClassLoader): Unit = {
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
index dfc650e..766ea55 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
@@ -22,10 +22,10 @@ import akka.actor.Status.Success
import akka.actor.{ActorRef, PoisonPill, ActorSystem}
import akka.testkit.{ImplicitSender, TestKit}
import org.apache.flink.configuration.{ConfigConstants, Configuration}
-import org.apache.flink.runtime.jobgraph.{JobStatus, JobGraph, DistributionPattern,AbstractJobVertex}
+import org.apache.flink.runtime.jobgraph.{JobStatus, JobGraph, DistributionPattern, JobVertex}
import org.apache.flink.runtime.jobmanager.Tasks.{BlockingOnceReceiver, FailingOnceReceiver}
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup
-import org.apache.flink.runtime.messages.JobManagerMessages.{JobResultSuccess, SubmitJob}
+import org.apache.flink.runtime.messages.JobManagerMessages.{ JobResultSuccess, SubmitJob}
import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages._
import org.apache.flink.runtime.testingUtils.{TestingCluster, TestingUtils}
import org.junit.runner.RunWith
@@ -59,8 +59,8 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
"recover once failing forward job" in {
FailingOnceReceiver.failed = false
- val sender = new AbstractJobVertex("Sender")
- val receiver = new AbstractJobVertex("Receiver")
+ val sender = new JobVertex("Sender")
+ val receiver = new JobVertex("Receiver")
sender.setInvokableClass(classOf[Tasks.Sender])
receiver.setInvokableClass(classOf[Tasks.FailingOnceReceiver])
@@ -98,8 +98,8 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
"recover once failing forward job with slot sharing" in {
FailingOnceReceiver.failed = false
- val sender = new AbstractJobVertex("Sender")
- val receiver = new AbstractJobVertex("Receiver")
+ val sender = new JobVertex("Sender")
+ val receiver = new JobVertex("Receiver")
sender.setInvokableClass(classOf[Tasks.Sender])
receiver.setInvokableClass(classOf[Tasks.FailingOnceReceiver])
@@ -141,8 +141,8 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
"recover a task manager failure" in {
BlockingOnceReceiver.blocking = true
- val sender = new AbstractJobVertex("Sender")
- val receiver = new AbstractJobVertex("Receiver")
+ val sender = new JobVertex("Sender")
+ val receiver = new JobVertex("Receiver")
sender.setInvokableClass(classOf[Tasks.Sender])
receiver.setInvokableClass(classOf[Tasks.BlockingOnceReceiver])
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/SlotSharingITCase.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/SlotSharingITCase.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/SlotSharingITCase.scala
index faff2f2..48b7971 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/SlotSharingITCase.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/SlotSharingITCase.scala
@@ -21,7 +21,7 @@ package org.apache.flink.runtime.jobmanager
import akka.actor.ActorSystem
import akka.actor.Status.Success
import akka.testkit.{ImplicitSender, TestKit}
-import org.apache.flink.runtime.jobgraph.{AbstractJobVertex, DistributionPattern, JobGraph}
+import org.apache.flink.runtime.jobgraph.{JobVertex, DistributionPattern, JobGraph}
import org.apache.flink.runtime.jobmanager.Tasks.{Sender, AgnosticBinaryReceiver, Receiver}
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup
import org.apache.flink.runtime.messages.JobManagerMessages.{JobResultSuccess, SubmitJob}
@@ -43,8 +43,8 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
"support slot sharing for forward job" in {
val num_tasks = 31
- val sender = new AbstractJobVertex("Sender")
- val receiver = new AbstractJobVertex("Receiver")
+ val sender = new JobVertex("Sender")
+ val receiver = new JobVertex("Receiver")
sender.setInvokableClass(classOf[Sender])
receiver.setInvokableClass(classOf[Receiver])
@@ -82,9 +82,9 @@ WordSpecLike with Matchers with BeforeAndAfterAll {
"support jobs with two inputs and slot sharing" in {
val num_tasks = 11
- val sender1 = new AbstractJobVertex("Sender1")
- val sender2 = new AbstractJobVertex("Sender2")
- val receiver = new AbstractJobVertex("Receiver")
+ val sender1 = new JobVertex("Sender1")
+ val sender2 = new JobVertex("Sender2")
+ val receiver = new JobVertex("Receiver")
sender1.setInvokableClass(classOf[Sender])
sender2.setInvokableClass(classOf[Sender])
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/TaskManagerFailsWithSlotSharingITCase.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/TaskManagerFailsWithSlotSharingITCase.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/TaskManagerFailsWithSlotSharingITCase.scala
index e98fd98..0fd7e1b 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/TaskManagerFailsWithSlotSharingITCase.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/TaskManagerFailsWithSlotSharingITCase.scala
@@ -22,7 +22,7 @@ import akka.actor.Status.{Failure, Success}
import akka.actor.{Kill, ActorSystem, PoisonPill}
import akka.testkit.{ImplicitSender, TestKit}
import org.apache.flink.runtime.client.JobExecutionException
-import org.apache.flink.runtime.jobgraph.{AbstractJobVertex, DistributionPattern, JobGraph}
+import org.apache.flink.runtime.jobgraph.{JobVertex, DistributionPattern, JobGraph}
import org.apache.flink.runtime.jobmanager.Tasks.{BlockingReceiver, Sender}
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup
import org.apache.flink.runtime.messages.JobManagerMessages.SubmitJob
@@ -46,8 +46,8 @@ ImplicitSender with WordSpecLike with Matchers with BeforeAndAfterAll {
"handle gracefully failing task manager with slot sharing" in {
val num_tasks = 100
- val sender = new AbstractJobVertex("Sender")
- val receiver = new AbstractJobVertex("Receiver")
+ val sender = new JobVertex("Sender")
+ val receiver = new JobVertex("Receiver")
sender.setInvokableClass(classOf[Sender])
receiver.setInvokableClass(classOf[BlockingReceiver])
@@ -95,8 +95,8 @@ ImplicitSender with WordSpecLike with Matchers with BeforeAndAfterAll {
"handle hard failing task manager with slot sharing" in {
val num_tasks = 20
- val sender = new AbstractJobVertex("Sender")
- val receiver = new AbstractJobVertex("Receiver")
+ val sender = new JobVertex("Sender")
+ val receiver = new JobVertex("Receiver")
sender.setInvokableClass(classOf[Sender])
receiver.setInvokableClass(classOf[BlockingReceiver])
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
index 6a6e899..531fc71 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
@@ -29,7 +29,7 @@ import java.util.Map.Entry;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobVertexID;
@@ -55,7 +55,7 @@ public class StreamingJobGraphGenerator {
private StreamGraph streamGraph;
- private Map<Integer, AbstractJobVertex> jobVertices;
+ private Map<Integer, JobVertex> jobVertices;
private JobGraph jobGraph;
private Collection<Integer> builtVertices;
@@ -71,7 +71,7 @@ public class StreamingJobGraphGenerator {
}
private void init() {
- this.jobVertices = new HashMap<Integer, AbstractJobVertex>();
+ this.jobVertices = new HashMap<Integer, JobVertex>();
this.builtVertices = new HashSet<Integer>();
this.chainedConfigs = new HashMap<Integer, Map<Integer, StreamConfig>>();
this.vertexConfigs = new HashMap<Integer, StreamConfig>();
@@ -220,7 +220,7 @@ public class StreamingJobGraphGenerator {
private StreamConfig createProcessingVertex(Integer vertexID) {
- AbstractJobVertex jobVertex = new AbstractJobVertex(chainedNames.get(vertexID));
+ JobVertex jobVertex = new JobVertex(chainedNames.get(vertexID));
StreamNode vertex = streamGraph.getStreamNode(vertexID);
jobVertex.setInvokableClass(vertex.getJobVertexClass());
@@ -294,8 +294,8 @@ public class StreamingJobGraphGenerator {
Integer downStreamvertexID = edge.getTargetId();
- AbstractJobVertex headVertex = jobVertices.get(headOfChain);
- AbstractJobVertex downStreamVertex = jobVertices.get(downStreamvertexID);
+ JobVertex headVertex = jobVertices.get(headOfChain);
+ JobVertex downStreamVertex = jobVertices.get(downStreamvertexID);
StreamConfig downStreamConfig = new StreamConfig(downStreamVertex.getConfiguration());
@@ -342,7 +342,7 @@ public class StreamingJobGraphGenerator {
Map<Integer, SlotSharingGroup> slotSharingGroups = new HashMap<Integer, SlotSharingGroup>();
- for (Entry<Integer, AbstractJobVertex> entry : jobVertices.entrySet()) {
+ for (Entry<Integer, JobVertex> entry : jobVertices.entrySet()) {
int slotSharingID = streamGraph.getStreamNode(entry.getKey()).getSlotSharingID();
@@ -358,8 +358,8 @@ public class StreamingJobGraphGenerator {
for (StreamLoop loop : streamGraph.getStreamLoops()) {
CoLocationGroup ccg = new CoLocationGroup();
- AbstractJobVertex tail = jobVertices.get(loop.getSink().getId());
- AbstractJobVertex head = jobVertices.get(loop.getSource().getId());
+ JobVertex tail = jobVertices.get(loop.getSink().getId());
+ JobVertex head = jobVertices.get(loop.getSource().getId());
ccg.addVertex(head);
ccg.addVertex(tail);
tail.updateCoLocationGroup(ccg);
@@ -388,7 +388,7 @@ public class StreamingJobGraphGenerator {
List<JobVertexID> commitVertices = new ArrayList<JobVertexID>();
- for (AbstractJobVertex vertex : jobVertices.values()) {
+ for (JobVertex vertex : jobVertices.values()) {
if (vertex.isInputVertex()) {
triggerVertices.add(vertex.getID());
commitVertices.add(vertex.getID());
[3/3] flink git commit: [FLINK-2120][runtime] rename
AbstractJobVertex to JobVertex
Posted by mx...@apache.org.
[FLINK-2120][runtime] rename AbstractJobVertex to JobVertex
This closes #840.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/1bd0af73
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/1bd0af73
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/1bd0af73
Branch: refs/heads/master
Commit: 1bd0af73892c812e6d16c5661cf10a9e2c13b3e4
Parents: 20d7e5a
Author: mjsax <mj...@informatik.hu-berlin.de>
Authored: Mon Jun 15 22:12:47 2015 +0200
Committer: Maximilian Michels <mx...@apache.org>
Committed: Tue Jun 16 17:18:45 2015 +0200
----------------------------------------------------------------------
.../client/program/ClientConnectionTest.java | 4 +-
.../plantranslate/JobGraphGenerator.java | 102 ++---
.../plantranslate/TempInIterationsTest.java | 4 +-
.../runtime/executiongraph/ExecutionGraph.java | 6 +-
.../executiongraph/ExecutionJobVertex.java | 10 +-
.../runtime/jobgraph/AbstractJobVertex.java | 405 -------------------
.../runtime/jobgraph/InputFormatVertex.java | 2 +-
.../runtime/jobgraph/IntermediateDataSet.java | 10 +-
.../apache/flink/runtime/jobgraph/JobEdge.java | 8 +-
.../apache/flink/runtime/jobgraph/JobGraph.java | 38 +-
.../flink/runtime/jobgraph/JobVertex.java | 405 +++++++++++++++++++
.../runtime/jobgraph/OutputFormatVertex.java | 2 +-
.../jobgraph/tasks/AbstractInvokable.java | 4 +-
.../jobmanager/scheduler/CoLocationGroup.java | 12 +-
.../checkpoint/CoordinatorShutdownTest.java | 6 +-
.../executiongraph/AllVerticesIteratorTest.java | 10 +-
.../ExecutionGraphConstructionTest.java | 102 ++---
.../ExecutionGraphDeploymentTest.java | 38 +-
.../executiongraph/ExecutionGraphTestUtils.java | 4 +-
.../ExecutionStateProgressTest.java | 4 +-
.../executiongraph/LocalInputSplitsTest.java | 6 +-
.../executiongraph/PointwisePatternTest.java | 44 +-
.../TerminalStateDeadlockTest.java | 8 +-
.../VertexLocationConstraintTest.java | 18 +-
.../executiongraph/VertexSlotSharingTest.java | 14 +-
.../flink/runtime/instance/SharedSlotsTest.java | 12 +-
.../PartialConsumePipelinedResultTest.java | 6 +-
.../flink/runtime/jobgraph/JobGraphTest.java | 74 ++--
.../runtime/jobgraph/JobTaskVertexTest.java | 10 +-
.../runtime/jobmanager/JobManagerTest.java | 4 +-
.../flink/runtime/jobmanager/JobSubmitTest.java | 6 +-
.../SlotCountExceedingParallelismTest.java | 6 +-
.../scheduler/CoLocationConstraintTest.java | 8 +-
.../ScheduleOrUpdateConsumersTest.java | 8 +-
.../runtime/taskmanager/TaskCancelTest.java | 10 +-
.../ExecutionGraphRestartTest.scala | 6 +-
.../TaskManagerLossFailsTasksTest.scala | 4 +-
.../jobmanager/CoLocationConstraintITCase.scala | 7 +-
.../runtime/jobmanager/JobManagerITCase.scala | 59 ++-
.../runtime/jobmanager/RecoveryITCase.scala | 16 +-
.../runtime/jobmanager/SlotSharingITCase.scala | 12 +-
.../TaskManagerFailsWithSlotSharingITCase.scala | 10 +-
.../api/graph/StreamingJobGraphGenerator.java | 20 +-
.../apache/flink/streaming/api/IterateTest.java | 12 +-
.../streaming/api/graph/SlotAllocationTest.java | 4 +-
.../BroadcastVarsNepheleITCase.java | 8 +-
.../KMeansIterativeNepheleITCase.java | 26 +-
.../JobSubmissionFailsITCase.java | 8 +-
.../ConnectedComponentsNepheleITCase.java | 58 +--
.../IterationWithChainingNepheleITCase.java | 8 +-
.../test/iterative/nephele/JobGraphUtils.java | 12 +-
.../CustomCompensatableDanglingPageRank.java | 10 +-
...mpensatableDanglingPageRankWithCombiner.java | 10 +-
.../CompensatableDanglingPageRank.java | 10 +-
.../runtime/NetworkStackThroughputITCase.java | 10 +-
.../jobmanager/JobManagerFailsITCase.scala | 6 +-
.../taskmanager/TaskManagerFailsITCase.scala | 14 +-
57 files changed, 869 insertions(+), 871 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-clients/src/test/java/org/apache/flink/client/program/ClientConnectionTest.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/program/ClientConnectionTest.java b/flink-clients/src/test/java/org/apache/flink/client/program/ClientConnectionTest.java
index be6c19a..39b74a3 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/program/ClientConnectionTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/program/ClientConnectionTest.java
@@ -20,7 +20,7 @@ package org.apache.flink.client.program;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.net.NetUtils;
@@ -87,7 +87,7 @@ public class ClientConnectionTest {
config.setString(ConfigConstants.AKKA_LOOKUP_TIMEOUT, (CONNECT_TIMEOUT/1000) + " s");
try {
- AbstractJobVertex vertex = new AbstractJobVertex("Test Vertex");
+ JobVertex vertex = new JobVertex("Test Vertex");
vertex.setInvokableClass(TestInvokable.class);
final JobGraph jg = new JobGraph("Test Job", vertex);
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-optimizer/src/main/java/org/apache/flink/optimizer/plantranslate/JobGraphGenerator.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plantranslate/JobGraphGenerator.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plantranslate/JobGraphGenerator.java
index 2630019..109be20 100644
--- a/flink-optimizer/src/main/java/org/apache/flink/optimizer/plantranslate/JobGraphGenerator.java
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/plantranslate/JobGraphGenerator.java
@@ -55,7 +55,7 @@ import org.apache.flink.runtime.iterative.task.IterationHeadPactTask;
import org.apache.flink.runtime.iterative.task.IterationIntermediatePactTask;
import org.apache.flink.runtime.iterative.task.IterationSynchronizationSinkTask;
import org.apache.flink.runtime.iterative.task.IterationTailPactTask;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.InputFormatVertex;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -110,7 +110,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
// ------------------------------------------------------------------------
- private Map<PlanNode, AbstractJobVertex> vertices; // a map from optimizer nodes to job vertices
+ private Map<PlanNode, JobVertex> vertices; // a map from optimizer nodes to job vertices
private Map<PlanNode, TaskInChain> chainedTasks; // a map from optimizer nodes to job vertices
@@ -118,7 +118,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
private List<TaskInChain> chainedTasksInSequence;
- private List<AbstractJobVertex> auxVertices; // auxiliary vertices which are added during job graph generation
+ private List<JobVertex> auxVertices; // auxiliary vertices which are added during job graph generation
private final int defaultMaxFan;
@@ -157,10 +157,10 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
* @return JobGraph generated frmo the plan.
*/
public JobGraph compileJobGraph(OptimizedPlan program) {
- this.vertices = new HashMap<PlanNode, AbstractJobVertex>();
+ this.vertices = new HashMap<PlanNode, JobVertex>();
this.chainedTasks = new HashMap<PlanNode, TaskInChain>();
this.chainedTasksInSequence = new ArrayList<TaskInChain>();
- this.auxVertices = new ArrayList<AbstractJobVertex>();
+ this.auxVertices = new ArrayList<JobVertex>();
this.iterations = new HashMap<IterationPlanNode, IterationDescriptor>();
this.iterationStack = new ArrayList<IterationPlanNode>();
@@ -198,11 +198,11 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
graph.setAllowQueuedScheduling(false);
// add vertices to the graph
- for (AbstractJobVertex vertex : this.vertices.values()) {
+ for (JobVertex vertex : this.vertices.values()) {
graph.addVertex(vertex);
}
- for (AbstractJobVertex vertex : this.auxVertices) {
+ for (JobVertex vertex : this.auxVertices) {
graph.addVertex(vertex);
vertex.setSlotSharingGroup(sharingGroup);
}
@@ -251,7 +251,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
}
// the vertex to be created for the current node
- final AbstractJobVertex vertex;
+ final JobVertex vertex;
try {
if (node instanceof SinkPlanNode) {
vertex = createDataSinkVertex((SinkPlanNode) node);
@@ -322,7 +322,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
// we adjust the joins / cogroups that go into the solution set here
for (Channel c : node.getOutgoingChannels()) {
DualInputPlanNode target = (DualInputPlanNode) c.getTarget();
- AbstractJobVertex accessingVertex = this.vertices.get(target);
+ JobVertex accessingVertex = this.vertices.get(target);
TaskConfig conf = new TaskConfig(accessingVertex.getConfiguration());
int inputNum = c == target.getInput1() ? 0 : c == target.getInput2() ? 1 : -1;
@@ -438,7 +438,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
if (node instanceof WorksetIterationPlanNode) {
// connect the initial solution set
WorksetIterationPlanNode wsNode = (WorksetIterationPlanNode) node;
- AbstractJobVertex headVertex = this.iterations.get(wsNode).getHeadTask();
+ JobVertex headVertex = this.iterations.get(wsNode).getHeadTask();
TaskConfig headConfig = new TaskConfig(headVertex.getConfiguration());
int inputIndex = headConfig.getDriverStrategy().getNumInputs();
headConfig.setIterationHeadSolutionSetInputIndex(inputIndex);
@@ -448,7 +448,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
return;
}
- final AbstractJobVertex targetVertex = this.vertices.get(node);
+ final JobVertex targetVertex = this.vertices.get(node);
// --------- Main Path: Translation of channels ----------
@@ -480,7 +480,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
throw new CompilerException("Bug: Found a chained task with an input ship strategy other than FORWARD.");
}
- AbstractJobVertex container = chainedTask.getContainingVertex();
+ JobVertex container = chainedTask.getContainingVertex();
if (container == null) {
final PlanNode sourceNode = inConn.getSource();
@@ -526,7 +526,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
if (this.currentIteration != null) {
- AbstractJobVertex head = this.iterations.get(this.currentIteration).getHeadTask();
+ JobVertex head = this.iterations.get(this.currentIteration).getHeadTask();
// the head may still be null if we descend into the static parts first
if (head != null) {
targetVertex.setStrictlyCoLocatedWith(head);
@@ -580,7 +580,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
}
}
- private int translateChannel(Channel input, int inputIndex, AbstractJobVertex targetVertex,
+ private int translateChannel(Channel input, int inputIndex, JobVertex targetVertex,
TaskConfig targetVertexConfig, boolean isBroadcast) throws Exception
{
final PlanNode inputPlanNode = input.getSource();
@@ -656,7 +656,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
}
final PlanNode sourceNode = inConn.getSource();
- AbstractJobVertex sourceVertex = this.vertices.get(sourceNode);
+ JobVertex sourceVertex = this.vertices.get(sourceNode);
TaskConfig sourceVertexConfig;
if (sourceVertex == null) {
@@ -740,7 +740,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
// Methods for creating individual vertices
// ------------------------------------------------------------------------
- private AbstractJobVertex createSingleInputVertex(SingleInputPlanNode node) throws CompilerException {
+ private JobVertex createSingleInputVertex(SingleInputPlanNode node) throws CompilerException {
final String taskName = node.getNodeName();
final DriverStrategy ds = node.getDriverStrategy();
@@ -783,7 +783,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
}
}
- final AbstractJobVertex vertex;
+ final JobVertex vertex;
final TaskConfig config;
if (chaining) {
@@ -792,7 +792,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
this.chainedTasks.put(node, new TaskInChain(ds.getPushChainDriverClass(), config, taskName));
} else {
// create task vertex
- vertex = new AbstractJobVertex(taskName);
+ vertex = new JobVertex(taskName);
vertex.setInvokableClass((this.currentIteration != null && node.isOnDynamicPath()) ? IterationIntermediatePactTask.class : RegularPactTask.class);
config = new TaskConfig(vertex.getConfiguration());
@@ -813,10 +813,10 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
return vertex;
}
- private AbstractJobVertex createDualInputVertex(DualInputPlanNode node) throws CompilerException {
+ private JobVertex createDualInputVertex(DualInputPlanNode node) throws CompilerException {
final String taskName = node.getNodeName();
final DriverStrategy ds = node.getDriverStrategy();
- final AbstractJobVertex vertex = new AbstractJobVertex(taskName);
+ final JobVertex vertex = new JobVertex(taskName);
final TaskConfig config = new TaskConfig(vertex.getConfiguration());
vertex.setInvokableClass( (this.currentIteration != null && node.isOnDynamicPath()) ? IterationIntermediatePactTask.class : RegularPactTask.class);
@@ -857,7 +857,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
return vertex;
}
- private AbstractJobVertex createDataSinkVertex(SinkPlanNode node) throws CompilerException {
+ private JobVertex createDataSinkVertex(SinkPlanNode node) throws CompilerException {
final OutputFormatVertex vertex = new OutputFormatVertex(node.getNodeName());
final TaskConfig config = new TaskConfig(vertex.getConfiguration());
@@ -871,7 +871,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
return vertex;
}
- private AbstractJobVertex createBulkIterationHead(BulkPartialSolutionPlanNode pspn) {
+ private JobVertex createBulkIterationHead(BulkPartialSolutionPlanNode pspn) {
// get the bulk iteration that corresponds to this partial solution node
final BulkIterationPlanNode iteration = pspn.getContainingIterationNode();
@@ -902,12 +902,12 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
}
// create or adopt the head vertex
- final AbstractJobVertex toReturn;
- final AbstractJobVertex headVertex;
+ final JobVertex toReturn;
+ final JobVertex headVertex;
final TaskConfig headConfig;
if (merge) {
final PlanNode successor = pspn.getOutgoingChannels().get(0).getTarget();
- headVertex = (AbstractJobVertex) this.vertices.get(successor);
+ headVertex = (JobVertex) this.vertices.get(successor);
if (headVertex == null) {
throw new CompilerException(
@@ -922,7 +922,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
// instantiate the head vertex and give it a no-op driver as the driver strategy.
// everything else happens in the post visit, after the input (the initial partial solution)
// is connected.
- headVertex = new AbstractJobVertex("PartialSolution ("+iteration.getNodeName()+")");
+ headVertex = new JobVertex("PartialSolution ("+iteration.getNodeName()+")");
headVertex.setInvokableClass(IterationHeadPactTask.class);
headConfig = new TaskConfig(headVertex.getConfiguration());
headConfig.setDriver(NoOpDriver.class);
@@ -939,7 +939,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
return toReturn;
}
- private AbstractJobVertex createWorksetIterationHead(WorksetPlanNode wspn) {
+ private JobVertex createWorksetIterationHead(WorksetPlanNode wspn) {
// get the bulk iteration that corresponds to this partial solution node
final WorksetIterationPlanNode iteration = wspn.getContainingIterationNode();
@@ -970,12 +970,12 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
}
// create or adopt the head vertex
- final AbstractJobVertex toReturn;
- final AbstractJobVertex headVertex;
+ final JobVertex toReturn;
+ final JobVertex headVertex;
final TaskConfig headConfig;
if (merge) {
final PlanNode successor = wspn.getOutgoingChannels().get(0).getTarget();
- headVertex = (AbstractJobVertex) this.vertices.get(successor);
+ headVertex = (JobVertex) this.vertices.get(successor);
if (headVertex == null) {
throw new CompilerException(
@@ -990,7 +990,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
// instantiate the head vertex and give it a no-op driver as the driver strategy.
// everything else happens in the post visit, after the input (the initial partial solution)
// is connected.
- headVertex = new AbstractJobVertex("IterationHead("+iteration.getNodeName()+")");
+ headVertex = new JobVertex("IterationHead("+iteration.getNodeName()+")");
headVertex.setInvokableClass(IterationHeadPactTask.class);
headConfig = new TaskConfig(headVertex.getConfiguration());
headConfig.setDriver(NoOpDriver.class);
@@ -1045,8 +1045,8 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
* @throws CompilerException
*/
private DistributionPattern connectJobVertices(Channel channel, int inputNumber,
- final AbstractJobVertex sourceVertex, final TaskConfig sourceConfig,
- final AbstractJobVertex targetVertex, final TaskConfig targetConfig, boolean isBroadcast)
+ final JobVertex sourceVertex, final TaskConfig sourceConfig,
+ final JobVertex targetVertex, final TaskConfig targetConfig, boolean isBroadcast)
throws CompilerException
{
// ------------ connect the vertices to the job graph --------------
@@ -1187,7 +1187,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
private void finalizeBulkIteration(IterationDescriptor descr) {
final BulkIterationPlanNode bulkNode = (BulkIterationPlanNode) descr.getIterationNode();
- final AbstractJobVertex headVertex = descr.getHeadTask();
+ final JobVertex headVertex = descr.getHeadTask();
final TaskConfig headConfig = new TaskConfig(headVertex.getConfiguration());
final TaskConfig headFinalOutputConfig = descr.getHeadFinalResultConfig();
@@ -1208,7 +1208,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
headConfig.setRelativeBackChannelMemory(relativeMemForBackChannel);
// --------------------------- create the sync task ---------------------------
- final AbstractJobVertex sync = new AbstractJobVertex("Sync(" + bulkNode.getNodeName() + ")");
+ final JobVertex sync = new JobVertex("Sync(" + bulkNode.getNodeName() + ")");
sync.setInvokableClass(IterationSynchronizationSinkTask.class);
sync.setParallelism(1);
this.auxVertices.add(sync);
@@ -1232,14 +1232,14 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
final PlanNode rootOfStepFunction = bulkNode.getRootOfStepFunction();
final TaskConfig tailConfig;
- AbstractJobVertex rootOfStepFunctionVertex = (AbstractJobVertex) this.vertices.get(rootOfStepFunction);
+ JobVertex rootOfStepFunctionVertex = (JobVertex) this.vertices.get(rootOfStepFunction);
if (rootOfStepFunctionVertex == null) {
// last op is chained
final TaskInChain taskInChain = this.chainedTasks.get(rootOfStepFunction);
if (taskInChain == null) {
throw new CompilerException("Bug: Tail of step function not found as vertex or chained task.");
}
- rootOfStepFunctionVertex = (AbstractJobVertex) taskInChain.getContainingVertex();
+ rootOfStepFunctionVertex = (JobVertex) taskInChain.getContainingVertex();
// the fake channel is statically typed to pact record. no data is sent over this channel anyways.
tailConfig = taskInChain.getTaskConfig();
@@ -1262,7 +1262,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
final TaskConfig tailConfigOfTerminationCriterion;
// If we have a termination criterion and it is not an intermediate node
if(rootOfTerminationCriterion != null && rootOfTerminationCriterion.getOutgoingChannels().isEmpty()) {
- AbstractJobVertex rootOfTerminationCriterionVertex = (AbstractJobVertex) this.vertices.get(rootOfTerminationCriterion);
+ JobVertex rootOfTerminationCriterionVertex = (JobVertex) this.vertices.get(rootOfTerminationCriterion);
if (rootOfTerminationCriterionVertex == null) {
@@ -1271,7 +1271,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
if (taskInChain == null) {
throw new CompilerException("Bug: Tail of termination criterion not found as vertex or chained task.");
}
- rootOfTerminationCriterionVertex = (AbstractJobVertex) taskInChain.getContainingVertex();
+ rootOfTerminationCriterionVertex = (JobVertex) taskInChain.getContainingVertex();
// the fake channel is statically typed to pact record. no data is sent over this channel anyways.
tailConfigOfTerminationCriterion = taskInChain.getTaskConfig();
@@ -1312,7 +1312,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
private void finalizeWorksetIteration(IterationDescriptor descr) {
final WorksetIterationPlanNode iterNode = (WorksetIterationPlanNode) descr.getIterationNode();
- final AbstractJobVertex headVertex = descr.getHeadTask();
+ final JobVertex headVertex = descr.getHeadTask();
final TaskConfig headConfig = new TaskConfig(headVertex.getConfiguration());
final TaskConfig headFinalOutputConfig = descr.getHeadFinalResultConfig();
@@ -1344,7 +1344,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
// --------------------------- create the sync task ---------------------------
final TaskConfig syncConfig;
{
- final AbstractJobVertex sync = new AbstractJobVertex("Sync (" + iterNode.getNodeName() + ")");
+ final JobVertex sync = new JobVertex("Sync (" + iterNode.getNodeName() + ")");
sync.setInvokableClass(IterationSynchronizationSinkTask.class);
sync.setParallelism(1);
this.auxVertices.add(sync);
@@ -1381,14 +1381,14 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
{
// get the vertex for the workset update
final TaskConfig worksetTailConfig;
- AbstractJobVertex nextWorksetVertex = (AbstractJobVertex) this.vertices.get(nextWorksetNode);
+ JobVertex nextWorksetVertex = (JobVertex) this.vertices.get(nextWorksetNode);
if (nextWorksetVertex == null) {
// nextWorksetVertex is chained
TaskInChain taskInChain = this.chainedTasks.get(nextWorksetNode);
if (taskInChain == null) {
throw new CompilerException("Bug: Next workset node not found as vertex or chained task.");
}
- nextWorksetVertex = (AbstractJobVertex) taskInChain.getContainingVertex();
+ nextWorksetVertex = (JobVertex) taskInChain.getContainingVertex();
worksetTailConfig = taskInChain.getTaskConfig();
} else {
worksetTailConfig = new TaskConfig(nextWorksetVertex.getConfiguration());
@@ -1406,14 +1406,14 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
}
{
final TaskConfig solutionDeltaConfig;
- AbstractJobVertex solutionDeltaVertex = (AbstractJobVertex) this.vertices.get(solutionDeltaNode);
+ JobVertex solutionDeltaVertex = (JobVertex) this.vertices.get(solutionDeltaNode);
if (solutionDeltaVertex == null) {
// last op is chained
TaskInChain taskInChain = this.chainedTasks.get(solutionDeltaNode);
if (taskInChain == null) {
throw new CompilerException("Bug: Solution Set Delta not found as vertex or chained task.");
}
- solutionDeltaVertex = (AbstractJobVertex) taskInChain.getContainingVertex();
+ solutionDeltaVertex = (JobVertex) taskInChain.getContainingVertex();
solutionDeltaConfig = taskInChain.getTaskConfig();
} else {
solutionDeltaConfig = new TaskConfig(solutionDeltaVertex.getConfiguration());
@@ -1501,7 +1501,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
private final String taskName;
- private AbstractJobVertex containingVertex;
+ private JobVertex containingVertex;
TaskInChain(Class<? extends ChainedDriver<?, ?>> chainedTask, TaskConfig taskConfig,
String taskName) {
@@ -1522,11 +1522,11 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
return this.taskName;
}
- public AbstractJobVertex getContainingVertex() {
+ public JobVertex getContainingVertex() {
return this.containingVertex;
}
- public void setContainingVertex(AbstractJobVertex containingVertex) {
+ public void setContainingVertex(JobVertex containingVertex) {
this.containingVertex = containingVertex;
}
}
@@ -1535,7 +1535,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
private final IterationPlanNode iterationNode;
- private AbstractJobVertex headTask;
+ private JobVertex headTask;
private TaskConfig headConfig;
@@ -1552,7 +1552,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
return iterationNode;
}
- public void setHeadTask(AbstractJobVertex headTask, TaskConfig headConfig) {
+ public void setHeadTask(JobVertex headTask, TaskConfig headConfig) {
this.headTask = headTask;
this.headFinalResultConfig = new TaskConfig(new Configuration());
@@ -1564,7 +1564,7 @@ public class JobGraphGenerator implements Visitor<PlanNode> {
this.headConfig = headConfig;
}
- public AbstractJobVertex getHeadTask() {
+ public JobVertex getHeadTask() {
return headTask;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-optimizer/src/test/java/org/apache/flink/optimizer/plantranslate/TempInIterationsTest.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/test/java/org/apache/flink/optimizer/plantranslate/TempInIterationsTest.java b/flink-optimizer/src/test/java/org/apache/flink/optimizer/plantranslate/TempInIterationsTest.java
index 15cb03f..a2a8d42 100644
--- a/flink-optimizer/src/test/java/org/apache/flink/optimizer/plantranslate/TempInIterationsTest.java
+++ b/flink-optimizer/src/test/java/org/apache/flink/optimizer/plantranslate/TempInIterationsTest.java
@@ -28,7 +28,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.optimizer.Optimizer;
import org.apache.flink.optimizer.plan.OptimizedPlan;
import org.apache.flink.optimizer.testfunctions.DummyFlatJoinFunction;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.operators.util.TaskConfig;
import org.junit.Test;
@@ -65,7 +65,7 @@ public class TempInIterationsTest {
JobGraph jg = jgg.compileJobGraph(oPlan);
boolean solutionSetUpdateChecked = false;
- for(AbstractJobVertex v : jg.getVertices()) {
+ for(JobVertex v : jg.getVertices()) {
if(v.getName().equals("SolutionSet Delta")) {
// check if input of solution set delta is temped
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
index 17eb3f6..84cbab7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
@@ -29,7 +29,7 @@ import org.apache.flink.runtime.blob.BlobKey;
import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.jobgraph.JobStatus;
@@ -455,7 +455,7 @@ public class ExecutionGraph implements Serializable {
// Actions
// --------------------------------------------------------------------------------------------
- public void attachJobGraph(List<AbstractJobVertex> topologiallySorted) throws JobException {
+ public void attachJobGraph(List<JobVertex> topologiallySorted) throws JobException {
if (LOG.isDebugEnabled()) {
LOG.debug(String.format("Attaching %d topologically sorted vertices to existing job graph with %d "
+ "vertices and %d intermediate results.", topologiallySorted.size(), tasks.size(), intermediateResults.size()));
@@ -463,7 +463,7 @@ public class ExecutionGraph implements Serializable {
final long createTimestamp = System.currentTimeMillis();
- for (AbstractJobVertex jobVertex : topologiallySorted) {
+ for (JobVertex jobVertex : topologiallySorted) {
// create the execution job vertex and attach it to the graph
ExecutionJobVertex ejv = new ExecutionJobVertex(this, jobVertex, 1, timeout, createTimestamp);
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/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 e53bc10..fcc0e9b 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
@@ -25,7 +25,7 @@ import org.apache.flink.core.io.InputSplitSource;
import org.apache.flink.core.io.LocatableInputSplit;
import org.apache.flink.runtime.JobException;
import org.apache.flink.runtime.instance.Instance;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.IntermediateDataSet;
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
import org.apache.flink.runtime.jobgraph.JobEdge;
@@ -57,7 +57,7 @@ public class ExecutionJobVertex implements Serializable {
private final ExecutionGraph graph;
- private final AbstractJobVertex jobVertex;
+ private final JobVertex jobVertex;
private final ExecutionVertex[] taskVertices;
@@ -81,12 +81,12 @@ public class ExecutionJobVertex implements Serializable {
private InputSplitAssigner splitAssigner;
- public ExecutionJobVertex(ExecutionGraph graph, AbstractJobVertex jobVertex,
+ public ExecutionJobVertex(ExecutionGraph graph, JobVertex jobVertex,
int defaultParallelism, FiniteDuration timeout) throws JobException {
this(graph, jobVertex, defaultParallelism, timeout, System.currentTimeMillis());
}
- public ExecutionJobVertex(ExecutionGraph graph, AbstractJobVertex jobVertex,
+ public ExecutionJobVertex(ExecutionGraph graph, JobVertex jobVertex,
int defaultParallelism, FiniteDuration timeout, long createTimestamp)
throws JobException
{
@@ -169,7 +169,7 @@ public class ExecutionJobVertex implements Serializable {
return graph;
}
- public AbstractJobVertex getJobVertex() {
+ public JobVertex getJobVertex() {
return jobVertex;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/AbstractJobVertex.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/AbstractJobVertex.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/AbstractJobVertex.java
deleted file mode 100644
index 63968b0..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/AbstractJobVertex.java
+++ /dev/null
@@ -1,405 +0,0 @@
-/*
- * 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.jobgraph;
-
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.core.io.InputSplitSource;
-import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
-import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
-import org.apache.flink.runtime.jobmanager.scheduler.CoLocationGroup;
-import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
-
-import com.google.common.base.Preconditions;
-
-/**
- * An abstract base class for a job vertex.
- */
-public class AbstractJobVertex implements java.io.Serializable {
-
- private static final long serialVersionUID = 1L;
-
- private static final String DEFAULT_NAME = "(unnamed vertex)";
-
-
- // --------------------------------------------------------------------------------------------
- // Members that define the structure / topology of the graph
- // --------------------------------------------------------------------------------------------
-
- /** The ID of the vertex. */
- private final JobVertexID id;
-
- /** List of produced data sets, one per writer */
- private final ArrayList<IntermediateDataSet> results = new ArrayList<IntermediateDataSet>();
-
- /** List of edges with incoming data. One per Reader. */
- private final ArrayList<JobEdge> inputs = new ArrayList<JobEdge>();
-
- /** Number of subtasks to split this task into at runtime.*/
- private int parallelism = -1;
-
- /** Custom configuration passed to the assigned task at runtime. */
- private Configuration configuration;
-
- /** The class of the invokable. */
- private String invokableClassName;
-
- /** Optionally, a source of input splits */
- private InputSplitSource<?> inputSplitSource;
-
- /** The name of the vertex */
- private String name;
-
- /** Optionally, a sharing group that allows subtasks from different job vertices to run concurrently in one slot */
- private SlotSharingGroup slotSharingGroup;
-
- /** The group inside which the vertex subtasks share slots */
- private CoLocationGroup coLocationGroup;
-
- // --------------------------------------------------------------------------------------------
-
- /**
- * Constructs a new job vertex and assigns it with the given name.
- *
- * @param name The name of the new job vertex.
- */
- public AbstractJobVertex(String name) {
- this(name, null);
- }
-
- /**
- * Constructs a new job vertex and assigns it with the given name.
- *
- * @param name The name of the new job vertex.
- * @param id The id of the job vertex.
- */
- public AbstractJobVertex(String name, JobVertexID id) {
- this.name = name == null ? DEFAULT_NAME : name;
- this.id = id == null ? new JobVertexID() : id;
- }
-
- // --------------------------------------------------------------------------------------------
-
- /**
- * Returns the ID of this job vertex.
- *
- * @return The ID of this job vertex
- */
- public JobVertexID getID() {
- return this.id;
- }
-
- /**
- * Returns the name of the vertex.
- *
- * @return The name of the vertex.
- */
- public String getName() {
- return this.name;
- }
-
- /**
- * Sets the name of the vertex
- *
- * @param name The new name.
- */
- public void setName(String name) {
- this.name = name == null ? DEFAULT_NAME : name;
- }
-
- /**
- * Returns the number of produced intermediate data sets.
- *
- * @return The number of produced intermediate data sets.
- */
- public int getNumberOfProducedIntermediateDataSets() {
- return this.results.size();
- }
-
- /**
- * Returns the number of inputs.
- *
- * @return The number of inputs.
- */
- public int getNumberOfInputs() {
- return this.inputs.size();
- }
-
- /**
- * Returns the vertex's configuration object which can be used to pass custom settings to the task at runtime.
- *
- * @return the vertex's configuration object
- */
- public Configuration getConfiguration() {
- if (this.configuration == null) {
- this.configuration = new Configuration();
- }
- return this.configuration;
- }
-
- public void setInvokableClass(Class<? extends AbstractInvokable> invokable) {
- Preconditions.checkNotNull(invokable);
- this.invokableClassName = invokable.getName();
- }
-
- /**
- * Returns the name of the invokable class which represents the task of this vertex.
- *
- * @return The name of the invokable class, <code>null</code> if not set.
- */
- public String getInvokableClassName() {
- return this.invokableClassName;
- }
-
- /**
- * Returns the invokable class which represents the task of this vertex
- *
- * @param cl The classloader used to resolve user-defined classes
- * @return The invokable class, <code>null</code> if it is not set
- */
- public Class<? extends AbstractInvokable> getInvokableClass(ClassLoader cl) {
- if (cl == null) {
- throw new NullPointerException("The classloader must not be null.");
- }
- if (invokableClassName == null) {
- return null;
- }
-
- try {
- return Class.forName(invokableClassName, true, cl).asSubclass(AbstractInvokable.class);
- }
- catch (ClassNotFoundException e) {
- throw new RuntimeException("The user-code class could not be resolved.", e);
- }
- catch (ClassCastException e) {
- throw new RuntimeException("The user-code class is no subclass of " + AbstractInvokable.class.getName(), e);
- }
- }
-
- /**
- * Gets the parallelism of the task.
- *
- * @return The parallelism of the task.
- */
- public int getParallelism() {
- return parallelism;
- }
-
- /**
- * Sets the parallelism for the task.
- *
- * @param parallelism The parallelism for the task.
- */
- public void setParallelism(int parallelism) {
- if (parallelism < 1) {
- throw new IllegalArgumentException("The parallelism must be at least one.");
- }
- this.parallelism = parallelism;
- }
-
- public InputSplitSource<?> getInputSplitSource() {
- return inputSplitSource;
- }
-
- public void setInputSplitSource(InputSplitSource<?> inputSplitSource) {
- this.inputSplitSource = inputSplitSource;
- }
-
- public List<IntermediateDataSet> getProducedDataSets() {
- return this.results;
- }
-
- public List<JobEdge> getInputs() {
- return this.inputs;
- }
-
- /**
- * Associates this vertex with a slot sharing group for scheduling. Different vertices in the same
- * slot sharing group can run one subtask each in the same slot.
- *
- * @param grp The slot sharing group to associate the vertex with.
- */
- public void setSlotSharingGroup(SlotSharingGroup grp) {
- if (this.slotSharingGroup != null) {
- this.slotSharingGroup.removeVertexFromGroup(id);
- }
-
- this.slotSharingGroup = grp;
- if (grp != null) {
- grp.addVertexToGroup(id);
- }
- }
-
- /**
- * Gets the slot sharing group that this vertex is associated with. Different vertices in the same
- * slot sharing group can run one subtask each in the same slot. If the vertex is not associated with
- * a slot sharing group, this method returns {@code null}.
- *
- * @return The slot sharing group to associate the vertex with, or {@code null}, if not associated with one.
- */
- public SlotSharingGroup getSlotSharingGroup() {
- return slotSharingGroup;
- }
-
- /**
- * Tells this vertex to strictly co locate its subtasks with the subtasks of the given vertex.
- * Strict co-location implies that the n'th subtask of this vertex will run on the same parallel computing
- * instance (TaskManager) as the n'th subtask of the given vertex.
- *
- * NOTE: Co-location is only possible between vertices in a slot sharing group.
- *
- * NOTE: This vertex must (transitively) depend on the vertex to be co-located with. That means that the
- * respective vertex must be a (transitive) input of this vertex.
- *
- * @param strictlyCoLocatedWith The vertex whose subtasks to co-locate this vertex's subtasks with.
- *
- * @throws IllegalArgumentException Thrown, if this vertex and the vertex to co-locate with are not in a common
- * slot sharing group.
- *
- * @see #setSlotSharingGroup(SlotSharingGroup)
- */
- public void setStrictlyCoLocatedWith(AbstractJobVertex strictlyCoLocatedWith) {
- if (this.slotSharingGroup == null || this.slotSharingGroup != strictlyCoLocatedWith.slotSharingGroup) {
- throw new IllegalArgumentException("Strict co-location requires that both vertices are in the same slot sharing group.");
- }
-
- CoLocationGroup thisGroup = this.coLocationGroup;
- CoLocationGroup otherGroup = strictlyCoLocatedWith.coLocationGroup;
-
- if (otherGroup == null) {
- if (thisGroup == null) {
- CoLocationGroup group = new CoLocationGroup(this, strictlyCoLocatedWith);
- this.coLocationGroup = group;
- strictlyCoLocatedWith.coLocationGroup = group;
- }
- else {
- thisGroup.addVertex(strictlyCoLocatedWith);
- strictlyCoLocatedWith.coLocationGroup = thisGroup;
- }
- }
- else {
- if (thisGroup == null) {
- otherGroup.addVertex(this);
- this.coLocationGroup = otherGroup;
- }
- else {
- // both had yet distinct groups, we need to merge them
- thisGroup.mergeInto(otherGroup);
- }
- }
- }
-
- public CoLocationGroup getCoLocationGroup() {
- return coLocationGroup;
- }
-
- public void updateCoLocationGroup(CoLocationGroup group) {
- this.coLocationGroup = group;
- }
-
- // --------------------------------------------------------------------------------------------
-
- public IntermediateDataSet createAndAddResultDataSet(ResultPartitionType partitionType) {
- return createAndAddResultDataSet(new IntermediateDataSetID(), partitionType);
- }
-
- public IntermediateDataSet createAndAddResultDataSet(
- IntermediateDataSetID id,
- ResultPartitionType partitionType) {
-
- IntermediateDataSet result = new IntermediateDataSet(id, partitionType, this);
- this.results.add(result);
- return result;
- }
-
- public void connectDataSetAsInput(IntermediateDataSet dataSet, DistributionPattern distPattern) {
- JobEdge edge = new JobEdge(dataSet, this, distPattern);
- this.inputs.add(edge);
- dataSet.addConsumer(edge);
- }
-
- public void connectNewDataSetAsInput(AbstractJobVertex input, DistributionPattern distPattern) {
- connectNewDataSetAsInput(input, distPattern, ResultPartitionType.PIPELINED);
- }
-
- public void connectNewDataSetAsInput(
- AbstractJobVertex input,
- DistributionPattern distPattern,
- ResultPartitionType partitionType) {
-
- IntermediateDataSet dataSet = input.createAndAddResultDataSet(partitionType);
- JobEdge edge = new JobEdge(dataSet, this, distPattern);
- this.inputs.add(edge);
- dataSet.addConsumer(edge);
- }
-
- public void connectIdInput(IntermediateDataSetID dataSetId, DistributionPattern distPattern) {
- JobEdge edge = new JobEdge(dataSetId, this, distPattern);
- this.inputs.add(edge);
- }
-
- // --------------------------------------------------------------------------------------------
-
- public boolean isInputVertex() {
- return this.inputs.isEmpty();
- }
-
- public boolean isOutputVertex() {
- return this.results.isEmpty();
- }
-
- public boolean hasNoConnectedInputs() {
- for (JobEdge edge : inputs) {
- if (!edge.isIdReference()) {
- return false;
- }
- }
-
- return true;
- }
-
- // --------------------------------------------------------------------------------------------
-
- /**
- * A hook that can be overwritten by sub classes to implement logic that is called by the
- * master when the job starts.
- *
- * @param loader The class loader for user defined code.
- * @throws Exception The method may throw exceptions which cause the job to fail immediately.
- */
- public void initializeOnMaster(ClassLoader loader) throws Exception {}
-
- /**
- * A hook that can be overwritten by sub classes to implement logic that is called by the
- * master after the job completed.
- *
- * @param loader The class loader for user defined code.
- * @throws Exception The method may throw exceptions which cause the job to fail immediately.
- */
- public void finalizeOnMaster(ClassLoader loader) throws Exception {}
-
- // --------------------------------------------------------------------------------------------
-
- @Override
- public String toString() {
- return this.name + " (" + this.invokableClassName + ')';
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/InputFormatVertex.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/InputFormatVertex.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/InputFormatVertex.java
index 0ea0da7..011850c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/InputFormatVertex.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/InputFormatVertex.java
@@ -22,7 +22,7 @@ import org.apache.flink.api.common.io.InputFormat;
import org.apache.flink.api.common.operators.util.UserCodeWrapper;
import org.apache.flink.runtime.operators.util.TaskConfig;
-public class InputFormatVertex extends AbstractJobVertex {
+public class InputFormatVertex extends JobVertex {
private static final long serialVersionUID = 1L;
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/IntermediateDataSet.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/IntermediateDataSet.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/IntermediateDataSet.java
index 86888e2..7c8f32b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/IntermediateDataSet.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/IntermediateDataSet.java
@@ -39,7 +39,7 @@ public class IntermediateDataSet implements java.io.Serializable {
private final IntermediateDataSetID id; // the identifier
- private final AbstractJobVertex producer; // the operation that produced this data set
+ private final JobVertex producer; // the operation that produced this data set
private final List<JobEdge> consumers = new ArrayList<JobEdge>();
@@ -48,15 +48,15 @@ public class IntermediateDataSet implements java.io.Serializable {
// --------------------------------------------------------------------------------------------
- public IntermediateDataSet(AbstractJobVertex producer) {
+ public IntermediateDataSet(JobVertex producer) {
this(new IntermediateDataSetID(), producer);
}
- public IntermediateDataSet(IntermediateDataSetID id, AbstractJobVertex producer) {
+ public IntermediateDataSet(IntermediateDataSetID id, JobVertex producer) {
this(id, ResultPartitionType.PIPELINED, producer);
}
- public IntermediateDataSet(IntermediateDataSetID id, ResultPartitionType resultType, AbstractJobVertex producer) {
+ public IntermediateDataSet(IntermediateDataSetID id, ResultPartitionType resultType, JobVertex producer) {
this.id = checkNotNull(id);
this.producer = checkNotNull(producer);
this.resultType = checkNotNull(resultType);
@@ -68,7 +68,7 @@ public class IntermediateDataSet implements java.io.Serializable {
return id;
}
- public AbstractJobVertex getProducer() {
+ public JobVertex getProducer() {
return producer;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobEdge.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobEdge.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobEdge.java
index 939f6c4..5faea83 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobEdge.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobEdge.java
@@ -29,7 +29,7 @@ public class JobEdge implements java.io.Serializable {
/** The vertex connected to this edge. */
- private final AbstractJobVertex target;
+ private final JobVertex target;
/** The distribution pattern that should be used for this job edge. */
private final DistributionPattern distributionPattern;
@@ -47,7 +47,7 @@ public class JobEdge implements java.io.Serializable {
* @param target The operation that is at the target of this edge.
* @param distributionPattern The pattern that defines how the connection behaves in parallel.
*/
- public JobEdge(IntermediateDataSet source, AbstractJobVertex target, DistributionPattern distributionPattern) {
+ public JobEdge(IntermediateDataSet source, JobVertex target, DistributionPattern distributionPattern) {
if (source == null || target == null || distributionPattern == null) {
throw new NullPointerException();
}
@@ -65,7 +65,7 @@ public class JobEdge implements java.io.Serializable {
* @param target The operation that is at the target of this edge.
* @param distributionPattern The pattern that defines how the connection behaves in parallel.
*/
- public JobEdge(IntermediateDataSetID sourceId, AbstractJobVertex target, DistributionPattern distributionPattern) {
+ public JobEdge(IntermediateDataSetID sourceId, JobVertex target, DistributionPattern distributionPattern) {
if (sourceId == null || target == null || distributionPattern == null) {
throw new NullPointerException();
}
@@ -90,7 +90,7 @@ public class JobEdge implements java.io.Serializable {
*
* @return The vertex connected to this edge.
*/
- public AbstractJobVertex getTarget() {
+ public JobVertex getTarget() {
return target;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
index 28fa78e..09b415b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
@@ -60,7 +60,7 @@ public class JobGraph implements Serializable {
// --------------------------------------------------------------------------------------------
/** List of task vertices included in this job graph. */
- private final Map<JobVertexID, AbstractJobVertex> taskVertices = new LinkedHashMap<JobVertexID, AbstractJobVertex>();
+ private final Map<JobVertexID, JobVertex> taskVertices = new LinkedHashMap<JobVertexID, JobVertex>();
/** The job configuration attached to this job. */
private final Configuration jobConfiguration = new Configuration();
@@ -124,7 +124,7 @@ public class JobGraph implements Serializable {
*
* @param vertices The vertices to add to the graph.
*/
- public JobGraph(AbstractJobVertex... vertices) {
+ public JobGraph(JobVertex... vertices) {
this(null, vertices);
}
@@ -134,7 +134,7 @@ public class JobGraph implements Serializable {
* @param jobName The name of the job.
* @param vertices The vertices to add to the graph.
*/
- public JobGraph(String jobName, AbstractJobVertex... vertices) {
+ public JobGraph(String jobName, JobVertex... vertices) {
this(null, jobName, vertices);
}
@@ -145,10 +145,10 @@ public class JobGraph implements Serializable {
* @param jobName The name of the job.
* @param vertices The vertices to add to the graph.
*/
- public JobGraph(JobID jobId, String jobName, AbstractJobVertex... vertices) {
+ public JobGraph(JobID jobId, String jobName, JobVertex... vertices) {
this(jobId, jobName);
- for (AbstractJobVertex vertex : vertices) {
+ for (JobVertex vertex : vertices) {
addVertex(vertex);
}
}
@@ -229,9 +229,9 @@ public class JobGraph implements Serializable {
* @param vertex
* the new task vertex to be added
*/
- public void addVertex(AbstractJobVertex vertex) {
+ public void addVertex(JobVertex vertex) {
final JobVertexID id = vertex.getID();
- AbstractJobVertex previous = taskVertices.put(id, vertex);
+ JobVertex previous = taskVertices.put(id, vertex);
// if we had a prior association, restore and throw an exception
if (previous != null) {
@@ -245,7 +245,7 @@ public class JobGraph implements Serializable {
*
* @return an Iterable to iterate all vertices registered with the job graph
*/
- public Iterable<AbstractJobVertex> getVertices() {
+ public Iterable<JobVertex> getVertices() {
return this.taskVertices.values();
}
@@ -255,8 +255,8 @@ public class JobGraph implements Serializable {
*
* @return an array of all job vertices that are registered with the job graph
*/
- public AbstractJobVertex[] getVerticesAsArray() {
- return this.taskVertices.values().toArray(new AbstractJobVertex[this.taskVertices.size()]);
+ public JobVertex[] getVerticesAsArray() {
+ return this.taskVertices.values().toArray(new JobVertex[this.taskVertices.size()]);
}
/**
@@ -295,27 +295,27 @@ public class JobGraph implements Serializable {
* the ID of the vertex to search for
* @return the vertex with the matching ID or <code>null</code> if no vertex with such ID could be found
*/
- public AbstractJobVertex findVertexByID(JobVertexID id) {
+ public JobVertex findVertexByID(JobVertexID id) {
return this.taskVertices.get(id);
}
// --------------------------------------------------------------------------------------------
- public List<AbstractJobVertex> getVerticesSortedTopologicallyFromSources() throws InvalidProgramException {
+ public List<JobVertex> getVerticesSortedTopologicallyFromSources() throws InvalidProgramException {
// early out on empty lists
if (this.taskVertices.isEmpty()) {
return Collections.emptyList();
}
- List<AbstractJobVertex> sorted = new ArrayList<AbstractJobVertex>(this.taskVertices.size());
- Set<AbstractJobVertex> remaining = new LinkedHashSet<AbstractJobVertex>(this.taskVertices.values());
+ List<JobVertex> sorted = new ArrayList<JobVertex>(this.taskVertices.size());
+ Set<JobVertex> remaining = new LinkedHashSet<JobVertex>(this.taskVertices.values());
// start by finding the vertices with no input edges
// and the ones with disconnected inputs (that refer to some standalone data set)
{
- Iterator<AbstractJobVertex> iter = remaining.iterator();
+ Iterator<JobVertex> iter = remaining.iterator();
while (iter.hasNext()) {
- AbstractJobVertex vertex = iter.next();
+ JobVertex vertex = iter.next();
if (vertex.hasNoConnectedInputs()) {
sorted.add(vertex);
@@ -335,21 +335,21 @@ public class JobGraph implements Serializable {
throw new InvalidProgramException("The job graph is cyclic.");
}
- AbstractJobVertex current = sorted.get(startNodePos++);
+ JobVertex current = sorted.get(startNodePos++);
addNodesThatHaveNoNewPredecessors(current, sorted, remaining);
}
return sorted;
}
- private void addNodesThatHaveNoNewPredecessors(AbstractJobVertex start, List<AbstractJobVertex> target, Set<AbstractJobVertex> remaining) {
+ private void addNodesThatHaveNoNewPredecessors(JobVertex start, List<JobVertex> target, Set<JobVertex> remaining) {
// forward traverse over all produced data sets and all their consumers
for (IntermediateDataSet dataSet : start.getProducedDataSets()) {
for (JobEdge edge : dataSet.getConsumers()) {
// a vertex can be added, if it has no predecessors that are still in the 'remaining' set
- AbstractJobVertex v = edge.getTarget();
+ JobVertex v = edge.getTarget();
if (!remaining.contains(v)) {
continue;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java
new file mode 100644
index 0000000..4bf9fc4
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java
@@ -0,0 +1,405 @@
+/*
+ * 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.jobgraph;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.io.InputSplitSource;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
+import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
+import org.apache.flink.runtime.jobmanager.scheduler.CoLocationGroup;
+import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * The base class for job vertexes.
+ */
+public class JobVertex implements java.io.Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final String DEFAULT_NAME = "(unnamed vertex)";
+
+
+ // --------------------------------------------------------------------------------------------
+ // Members that define the structure / topology of the graph
+ // --------------------------------------------------------------------------------------------
+
+ /** The ID of the vertex. */
+ private final JobVertexID id;
+
+ /** List of produced data sets, one per writer */
+ private final ArrayList<IntermediateDataSet> results = new ArrayList<IntermediateDataSet>();
+
+ /** List of edges with incoming data. One per Reader. */
+ private final ArrayList<JobEdge> inputs = new ArrayList<JobEdge>();
+
+ /** Number of subtasks to split this task into at runtime.*/
+ private int parallelism = -1;
+
+ /** Custom configuration passed to the assigned task at runtime. */
+ private Configuration configuration;
+
+ /** The class of the invokable. */
+ private String invokableClassName;
+
+ /** Optionally, a source of input splits */
+ private InputSplitSource<?> inputSplitSource;
+
+ /** The name of the vertex */
+ private String name;
+
+ /** Optionally, a sharing group that allows subtasks from different job vertices to run concurrently in one slot */
+ private SlotSharingGroup slotSharingGroup;
+
+ /** The group inside which the vertex subtasks share slots */
+ private CoLocationGroup coLocationGroup;
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Constructs a new job vertex and assigns it with the given name.
+ *
+ * @param name The name of the new job vertex.
+ */
+ public JobVertex(String name) {
+ this(name, null);
+ }
+
+ /**
+ * Constructs a new job vertex and assigns it with the given name.
+ *
+ * @param name The name of the new job vertex.
+ * @param id The id of the job vertex.
+ */
+ public JobVertex(String name, JobVertexID id) {
+ this.name = name == null ? DEFAULT_NAME : name;
+ this.id = id == null ? new JobVertexID() : id;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Returns the ID of this job vertex.
+ *
+ * @return The ID of this job vertex
+ */
+ public JobVertexID getID() {
+ return this.id;
+ }
+
+ /**
+ * Returns the name of the vertex.
+ *
+ * @return The name of the vertex.
+ */
+ public String getName() {
+ return this.name;
+ }
+
+ /**
+ * Sets the name of the vertex
+ *
+ * @param name The new name.
+ */
+ public void setName(String name) {
+ this.name = name == null ? DEFAULT_NAME : name;
+ }
+
+ /**
+ * Returns the number of produced intermediate data sets.
+ *
+ * @return The number of produced intermediate data sets.
+ */
+ public int getNumberOfProducedIntermediateDataSets() {
+ return this.results.size();
+ }
+
+ /**
+ * Returns the number of inputs.
+ *
+ * @return The number of inputs.
+ */
+ public int getNumberOfInputs() {
+ return this.inputs.size();
+ }
+
+ /**
+ * Returns the vertex's configuration object which can be used to pass custom settings to the task at runtime.
+ *
+ * @return the vertex's configuration object
+ */
+ public Configuration getConfiguration() {
+ if (this.configuration == null) {
+ this.configuration = new Configuration();
+ }
+ return this.configuration;
+ }
+
+ public void setInvokableClass(Class<? extends AbstractInvokable> invokable) {
+ Preconditions.checkNotNull(invokable);
+ this.invokableClassName = invokable.getName();
+ }
+
+ /**
+ * Returns the name of the invokable class which represents the task of this vertex.
+ *
+ * @return The name of the invokable class, <code>null</code> if not set.
+ */
+ public String getInvokableClassName() {
+ return this.invokableClassName;
+ }
+
+ /**
+ * Returns the invokable class which represents the task of this vertex
+ *
+ * @param cl The classloader used to resolve user-defined classes
+ * @return The invokable class, <code>null</code> if it is not set
+ */
+ public Class<? extends AbstractInvokable> getInvokableClass(ClassLoader cl) {
+ if (cl == null) {
+ throw new NullPointerException("The classloader must not be null.");
+ }
+ if (invokableClassName == null) {
+ return null;
+ }
+
+ try {
+ return Class.forName(invokableClassName, true, cl).asSubclass(AbstractInvokable.class);
+ }
+ catch (ClassNotFoundException e) {
+ throw new RuntimeException("The user-code class could not be resolved.", e);
+ }
+ catch (ClassCastException e) {
+ throw new RuntimeException("The user-code class is no subclass of " + AbstractInvokable.class.getName(), e);
+ }
+ }
+
+ /**
+ * Gets the parallelism of the task.
+ *
+ * @return The parallelism of the task.
+ */
+ public int getParallelism() {
+ return parallelism;
+ }
+
+ /**
+ * Sets the parallelism for the task.
+ *
+ * @param parallelism The parallelism for the task.
+ */
+ public void setParallelism(int parallelism) {
+ if (parallelism < 1) {
+ throw new IllegalArgumentException("The parallelism must be at least one.");
+ }
+ this.parallelism = parallelism;
+ }
+
+ public InputSplitSource<?> getInputSplitSource() {
+ return inputSplitSource;
+ }
+
+ public void setInputSplitSource(InputSplitSource<?> inputSplitSource) {
+ this.inputSplitSource = inputSplitSource;
+ }
+
+ public List<IntermediateDataSet> getProducedDataSets() {
+ return this.results;
+ }
+
+ public List<JobEdge> getInputs() {
+ return this.inputs;
+ }
+
+ /**
+ * Associates this vertex with a slot sharing group for scheduling. Different vertices in the same
+ * slot sharing group can run one subtask each in the same slot.
+ *
+ * @param grp The slot sharing group to associate the vertex with.
+ */
+ public void setSlotSharingGroup(SlotSharingGroup grp) {
+ if (this.slotSharingGroup != null) {
+ this.slotSharingGroup.removeVertexFromGroup(id);
+ }
+
+ this.slotSharingGroup = grp;
+ if (grp != null) {
+ grp.addVertexToGroup(id);
+ }
+ }
+
+ /**
+ * Gets the slot sharing group that this vertex is associated with. Different vertices in the same
+ * slot sharing group can run one subtask each in the same slot. If the vertex is not associated with
+ * a slot sharing group, this method returns {@code null}.
+ *
+ * @return The slot sharing group to associate the vertex with, or {@code null}, if not associated with one.
+ */
+ public SlotSharingGroup getSlotSharingGroup() {
+ return slotSharingGroup;
+ }
+
+ /**
+ * Tells this vertex to strictly co locate its subtasks with the subtasks of the given vertex.
+ * Strict co-location implies that the n'th subtask of this vertex will run on the same parallel computing
+ * instance (TaskManager) as the n'th subtask of the given vertex.
+ *
+ * NOTE: Co-location is only possible between vertices in a slot sharing group.
+ *
+ * NOTE: This vertex must (transitively) depend on the vertex to be co-located with. That means that the
+ * respective vertex must be a (transitive) input of this vertex.
+ *
+ * @param strictlyCoLocatedWith The vertex whose subtasks to co-locate this vertex's subtasks with.
+ *
+ * @throws IllegalArgumentException Thrown, if this vertex and the vertex to co-locate with are not in a common
+ * slot sharing group.
+ *
+ * @see #setSlotSharingGroup(SlotSharingGroup)
+ */
+ public void setStrictlyCoLocatedWith(JobVertex strictlyCoLocatedWith) {
+ if (this.slotSharingGroup == null || this.slotSharingGroup != strictlyCoLocatedWith.slotSharingGroup) {
+ throw new IllegalArgumentException("Strict co-location requires that both vertices are in the same slot sharing group.");
+ }
+
+ CoLocationGroup thisGroup = this.coLocationGroup;
+ CoLocationGroup otherGroup = strictlyCoLocatedWith.coLocationGroup;
+
+ if (otherGroup == null) {
+ if (thisGroup == null) {
+ CoLocationGroup group = new CoLocationGroup(this, strictlyCoLocatedWith);
+ this.coLocationGroup = group;
+ strictlyCoLocatedWith.coLocationGroup = group;
+ }
+ else {
+ thisGroup.addVertex(strictlyCoLocatedWith);
+ strictlyCoLocatedWith.coLocationGroup = thisGroup;
+ }
+ }
+ else {
+ if (thisGroup == null) {
+ otherGroup.addVertex(this);
+ this.coLocationGroup = otherGroup;
+ }
+ else {
+ // both had yet distinct groups, we need to merge them
+ thisGroup.mergeInto(otherGroup);
+ }
+ }
+ }
+
+ public CoLocationGroup getCoLocationGroup() {
+ return coLocationGroup;
+ }
+
+ public void updateCoLocationGroup(CoLocationGroup group) {
+ this.coLocationGroup = group;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ public IntermediateDataSet createAndAddResultDataSet(ResultPartitionType partitionType) {
+ return createAndAddResultDataSet(new IntermediateDataSetID(), partitionType);
+ }
+
+ public IntermediateDataSet createAndAddResultDataSet(
+ IntermediateDataSetID id,
+ ResultPartitionType partitionType) {
+
+ IntermediateDataSet result = new IntermediateDataSet(id, partitionType, this);
+ this.results.add(result);
+ return result;
+ }
+
+ public void connectDataSetAsInput(IntermediateDataSet dataSet, DistributionPattern distPattern) {
+ JobEdge edge = new JobEdge(dataSet, this, distPattern);
+ this.inputs.add(edge);
+ dataSet.addConsumer(edge);
+ }
+
+ public void connectNewDataSetAsInput(JobVertex input, DistributionPattern distPattern) {
+ connectNewDataSetAsInput(input, distPattern, ResultPartitionType.PIPELINED);
+ }
+
+ public void connectNewDataSetAsInput(
+ JobVertex input,
+ DistributionPattern distPattern,
+ ResultPartitionType partitionType) {
+
+ IntermediateDataSet dataSet = input.createAndAddResultDataSet(partitionType);
+ JobEdge edge = new JobEdge(dataSet, this, distPattern);
+ this.inputs.add(edge);
+ dataSet.addConsumer(edge);
+ }
+
+ public void connectIdInput(IntermediateDataSetID dataSetId, DistributionPattern distPattern) {
+ JobEdge edge = new JobEdge(dataSetId, this, distPattern);
+ this.inputs.add(edge);
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ public boolean isInputVertex() {
+ return this.inputs.isEmpty();
+ }
+
+ public boolean isOutputVertex() {
+ return this.results.isEmpty();
+ }
+
+ public boolean hasNoConnectedInputs() {
+ for (JobEdge edge : inputs) {
+ if (!edge.isIdReference()) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * A hook that can be overwritten by sub classes to implement logic that is called by the
+ * master when the job starts.
+ *
+ * @param loader The class loader for user defined code.
+ * @throws Exception The method may throw exceptions which cause the job to fail immediately.
+ */
+ public void initializeOnMaster(ClassLoader loader) throws Exception {}
+
+ /**
+ * A hook that can be overwritten by sub classes to implement logic that is called by the
+ * master after the job completed.
+ *
+ * @param loader The class loader for user defined code.
+ * @throws Exception The method may throw exceptions which cause the job to fail immediately.
+ */
+ public void finalizeOnMaster(ClassLoader loader) throws Exception {}
+
+ // --------------------------------------------------------------------------------------------
+
+ @Override
+ public String toString() {
+ return this.name + " (" + this.invokableClassName + ')';
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/OutputFormatVertex.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/OutputFormatVertex.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/OutputFormatVertex.java
index 2a1f89c..c9ac564 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/OutputFormatVertex.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/OutputFormatVertex.java
@@ -28,7 +28,7 @@ import org.apache.flink.runtime.operators.util.TaskConfig;
* A task vertex that run an initialization on the master, trying to deserialize an output format
* and initializing it on master, if necessary.
*/
-public class OutputFormatVertex extends AbstractJobVertex {
+public class OutputFormatVertex extends JobVertex {
private static final long serialVersionUID = 1L;
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java
index a70cd2b..85dd5c5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java
@@ -108,9 +108,9 @@ public abstract class AbstractInvokable {
}
/**
- * Returns the task configuration object which was attached to the original {@link org.apache.flink.runtime.jobgraph.AbstractJobVertex}.
+ * Returns the task configuration object which was attached to the original {@link org.apache.flink.runtime.jobgraph.JobVertex}.
*
- * @return the task configuration object which was attached to the original {@link org.apache.flink.runtime.jobgraph.AbstractJobVertex}
+ * @return the task configuration object which was attached to the original {@link org.apache.flink.runtime.jobgraph.JobVertex}
*/
public Configuration getTaskConfiguration() {
return this.environment.getTaskConfiguration();
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroup.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroup.java
index a88c89d..74e34e1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroup.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroup.java
@@ -22,7 +22,7 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.flink.util.AbstractID;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import com.google.common.base.Preconditions;
@@ -43,7 +43,7 @@ public class CoLocationGroup implements java.io.Serializable {
private final AbstractID id = new AbstractID();
/** The vertices participating in the co-location group */
- private final List<AbstractJobVertex> vertices = new ArrayList<AbstractJobVertex>();
+ private final List<JobVertex> vertices = new ArrayList<JobVertex>();
/** The constraints, which hold the shared slots for the co-located operators */
private transient ArrayList<CoLocationConstraint> constraints;
@@ -52,15 +52,15 @@ public class CoLocationGroup implements java.io.Serializable {
public CoLocationGroup() {}
- public CoLocationGroup(AbstractJobVertex... vertices) {
- for (AbstractJobVertex v : vertices) {
+ public CoLocationGroup(JobVertex... vertices) {
+ for (JobVertex v : vertices) {
this.vertices.add(v);
}
}
// --------------------------------------------------------------------------------------------
- public void addVertex(AbstractJobVertex vertex) {
+ public void addVertex(JobVertex vertex) {
Preconditions.checkNotNull(vertex);
this.vertices.add(vertex);
}
@@ -68,7 +68,7 @@ public class CoLocationGroup implements java.io.Serializable {
public void mergeInto(CoLocationGroup other) {
Preconditions.checkNotNull(other);
- for (AbstractJobVertex v : this.vertices) {
+ for (JobVertex v : this.vertices) {
v.updateCoLocationGroup(other);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CoordinatorShutdownTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CoordinatorShutdownTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CoordinatorShutdownTest.java
index 3d9a155..932e366 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CoordinatorShutdownTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CoordinatorShutdownTest.java
@@ -24,7 +24,7 @@ import akka.pattern.Patterns;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.executiongraph.ExecutionGraph;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.JobSnapshottingSettings;
@@ -55,7 +55,7 @@ public class CoordinatorShutdownTest {
cluster = new LocalFlinkMiniCluster(noTaskManagerConfig, true);
// build a test graph with snapshotting enabled
- AbstractJobVertex vertex = new AbstractJobVertex("Test Vertex");
+ JobVertex vertex = new JobVertex("Test Vertex");
vertex.setInvokableClass(Tasks.NoOpInvokable.class);
List<JobVertexID> vertexIdList = Collections.singletonList(vertex.getID());
@@ -102,7 +102,7 @@ public class CoordinatorShutdownTest {
cluster = new LocalFlinkMiniCluster(new Configuration(), true);
// build a test graph with snapshotting enabled
- AbstractJobVertex vertex = new AbstractJobVertex("Test Vertex");
+ JobVertex vertex = new JobVertex("Test Vertex");
vertex.setInvokableClass(Tasks.NoOpInvokable.class);
List<JobVertexID> vertexIdList = Collections.singletonList(vertex.getID());
http://git-wip-us.apache.org/repos/asf/flink/blob/1bd0af73/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/AllVerticesIteratorTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/AllVerticesIteratorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/AllVerticesIteratorTest.java
index 4d10585..693e014 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/AllVerticesIteratorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/AllVerticesIteratorTest.java
@@ -21,7 +21,7 @@ package org.apache.flink.runtime.executiongraph;
import java.util.Arrays;
import org.apache.flink.runtime.akka.AkkaUtils;
-import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertex;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
@@ -32,10 +32,10 @@ public class AllVerticesIteratorTest {
public void testAllVertices() {
try {
- AbstractJobVertex v1 = new AbstractJobVertex("v1");
- AbstractJobVertex v2 = new AbstractJobVertex("v2");
- AbstractJobVertex v3 = new AbstractJobVertex("v3");
- AbstractJobVertex v4 = new AbstractJobVertex("v4");
+ JobVertex v1 = new JobVertex("v1");
+ JobVertex v2 = new JobVertex("v2");
+ JobVertex v3 = new JobVertex("v3");
+ JobVertex v4 = new JobVertex("v4");
v1.setParallelism(1);
v2.setParallelism(7);