You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/01/10 12:46:06 UTC
[2/6] flink git commit: [FLINK-7910] [tests] Generalize
Test(Stream)Environment to use JobExecutor
[FLINK-7910] [tests] Generalize Test(Stream)Environment to use JobExecutor
This commit introduces the JobExecutor interface which abstracts the actual mini cluster
from the Test(Stream)Environment. By letting the Flip-6 MiniCluster as well as the
FlinkMiniCluster implement this interface, we can run all test base jobs either on the
Flip-6 mini cluster or on the current mini cluster.
This closes #4897.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/057edf9e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/057edf9e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/057edf9e
Branch: refs/heads/master
Commit: 057edf9e28b656401b985069ebcc428ab55e5fed
Parents: 51a2787
Author: Till Rohrmann <tr...@apache.org>
Authored: Mon Jan 8 18:23:27 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Wed Jan 10 12:30:30 2018 +0100
----------------------------------------------------------------------
.../flink/runtime/minicluster/JobExecutor.java | 38 ++++++++++++++++++++
.../flink/runtime/minicluster/MiniCluster.java | 5 +--
.../runtime/minicluster/FlinkMiniCluster.scala | 14 +++++++-
.../runtime/minicluster/MiniClusterITCase.java | 2 +-
.../Flip6LocalStreamEnvironment.java | 2 +-
.../streaming/util/TestStreamEnvironment.java | 31 ++++++++--------
.../apache/flink/test/util/TestEnvironment.java | 36 ++++++++++---------
7 files changed, 91 insertions(+), 37 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/057edf9e/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/JobExecutor.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/JobExecutor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/JobExecutor.java
new file mode 100644
index 0000000..4991c2f
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/JobExecutor.java
@@ -0,0 +1,38 @@
+/*
+ * 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.minicluster;
+
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.runtime.client.JobExecutionException;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+
+/**
+ * Interface for {@link JobGraph} executors.
+ */
+public interface JobExecutor {
+
+ /**
+ * Run the given job and block until its execution result can be returned.
+ *
+ * @param jobGraph to execute
+ * @return Execution result of the executed job
+ * @throws JobExecutionException if the job failed to execute
+ */
+ JobExecutionResult executeJobBlocking(final JobGraph jobGraph) throws JobExecutionException, InterruptedException;
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/057edf9e/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
index 2bbd2c7..2598a60 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java
@@ -58,7 +58,7 @@ import static org.apache.flink.util.ExceptionUtils.firstOrSuppressed;
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;
-public class MiniCluster {
+public class MiniCluster implements JobExecutor {
private static final Logger LOG = LoggerFactory.getLogger(MiniCluster.class);
@@ -448,7 +448,8 @@ public class MiniCluster {
* @throws JobExecutionException Thrown if anything went amiss during initial job launch,
* or if the job terminally failed.
*/
- public JobExecutionResult runJobBlocking(JobGraph job) throws JobExecutionException, InterruptedException {
+ @Override
+ public JobExecutionResult executeJobBlocking(JobGraph job) throws JobExecutionException, InterruptedException {
checkNotNull(job, "job is null");
MiniClusterJobDispatcher dispatcher;
http://git-wip-us.apache.org/repos/asf/flink/blob/057edf9e/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala
index 5554061..cc8ae5f 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala
@@ -64,7 +64,8 @@ abstract class FlinkMiniCluster(
val userConfiguration: Configuration,
val highAvailabilityServices: HighAvailabilityServices,
val useSingleActorSystem: Boolean)
- extends LeaderRetrievalListener {
+ extends LeaderRetrievalListener
+ with JobExecutor {
protected val LOG = LoggerFactory.getLogger(classOf[FlinkMiniCluster])
@@ -701,4 +702,15 @@ abstract class FlinkMiniCluster(
FlinkUserCodeClassLoaders.parentFirst(urls, parentClassLoader)
}
+
+ /**
+ * Run the given job and block until its execution result can be returned.
+ *
+ * @param jobGraph to execute
+ * @return Execution result of the executed job
+ * @throws JobExecutionException if the job failed to execute
+ */
+ override def executeJobBlocking(jobGraph: JobGraph) = {
+ submitJobAndWait(jobGraph, false)
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/057edf9e/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/MiniClusterITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/MiniClusterITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/MiniClusterITCase.java
index f822e0c..6a4f678 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/MiniClusterITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/MiniClusterITCase.java
@@ -99,7 +99,7 @@ public class MiniClusterITCase extends TestLogger {
private static void executeJob(MiniCluster miniCluster) throws Exception {
JobGraph job = getSimpleJob();
- miniCluster.runJobBlocking(job);
+ miniCluster.executeJobBlocking(job);
}
private static JobGraph getSimpleJob() throws IOException {
http://git-wip-us.apache.org/repos/asf/flink/blob/057edf9e/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/Flip6LocalStreamEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/Flip6LocalStreamEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/Flip6LocalStreamEnvironment.java
index 8720e7a..9d4f2a7 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/Flip6LocalStreamEnvironment.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/Flip6LocalStreamEnvironment.java
@@ -107,7 +107,7 @@ public class Flip6LocalStreamEnvironment extends StreamExecutionEnvironment {
try {
miniCluster.start();
- return miniCluster.runJobBlocking(jobGraph);
+ return miniCluster.executeJobBlocking(jobGraph);
}
finally {
transformations.clear();
http://git-wip-us.apache.org/repos/asf/flink/blob/057edf9e/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/streaming/util/TestStreamEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/streaming/util/TestStreamEnvironment.java b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/streaming/util/TestStreamEnvironment.java
index 86159a5..61615b8 100644
--- a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/streaming/util/TestStreamEnvironment.java
+++ b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/streaming/util/TestStreamEnvironment.java
@@ -21,6 +21,7 @@ package org.apache.flink.streaming.util;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.minicluster.JobExecutor;
import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironmentFactory;
@@ -37,20 +38,20 @@ import java.util.Collections;
*/
public class TestStreamEnvironment extends StreamExecutionEnvironment {
- /** The mini cluster in which this environment executes its jobs. */
- private final LocalFlinkMiniCluster miniCluster;
+ /** The job executor to use to execute environment's jobs. */
+ private final JobExecutor jobExecutor;
private final Collection<Path> jarFiles;
private final Collection<URL> classPaths;
public TestStreamEnvironment(
- LocalFlinkMiniCluster miniCluster,
+ JobExecutor jobExecutor,
int parallelism,
Collection<Path> jarFiles,
Collection<URL> classPaths) {
- this.miniCluster = Preconditions.checkNotNull(miniCluster);
+ this.jobExecutor = Preconditions.checkNotNull(jobExecutor);
this.jarFiles = Preconditions.checkNotNull(jarFiles);
this.classPaths = Preconditions.checkNotNull(classPaths);
@@ -58,9 +59,9 @@ public class TestStreamEnvironment extends StreamExecutionEnvironment {
}
public TestStreamEnvironment(
- LocalFlinkMiniCluster miniCluster,
+ JobExecutor jobExecutor,
int parallelism) {
- this(miniCluster, parallelism, Collections.<Path>emptyList(), Collections.<URL>emptyList());
+ this(jobExecutor, parallelism, Collections.emptyList(), Collections.emptyList());
}
@Override
@@ -75,7 +76,7 @@ public class TestStreamEnvironment extends StreamExecutionEnvironment {
jobGraph.setClasspaths(new ArrayList<>(classPaths));
- return miniCluster.submitJobAndWait(jobGraph, false);
+ return jobExecutor.executeJobBlocking(jobGraph);
}
// ------------------------------------------------------------------------
@@ -85,13 +86,13 @@ public class TestStreamEnvironment extends StreamExecutionEnvironment {
* the given cluster with the given default parallelism and the specified jar files and class
* paths.
*
- * @param cluster The test cluster to run the test program on.
+ * @param jobExecutor The executor to execute the jobs on
* @param parallelism The default parallelism for the test programs.
* @param jarFiles Additional jar files to execute the job with
* @param classpaths Additional class paths to execute the job with
*/
public static void setAsContext(
- final LocalFlinkMiniCluster cluster,
+ final JobExecutor jobExecutor,
final int parallelism,
final Collection<Path> jarFiles,
final Collection<URL> classpaths) {
@@ -100,7 +101,7 @@ public class TestStreamEnvironment extends StreamExecutionEnvironment {
@Override
public StreamExecutionEnvironment createExecutionEnvironment() {
return new TestStreamEnvironment(
- cluster,
+ jobExecutor,
parallelism,
jarFiles,
classpaths);
@@ -114,15 +115,15 @@ public class TestStreamEnvironment extends StreamExecutionEnvironment {
* Sets the streaming context environment to a TestStreamEnvironment that runs its programs on
* the given cluster with the given default parallelism.
*
- * @param cluster The test cluster to run the test program on.
+ * @param jobExecutor The executor to execute the jobs on
* @param parallelism The default parallelism for the test programs.
*/
- public static void setAsContext(final LocalFlinkMiniCluster cluster, final int parallelism) {
+ public static void setAsContext(final JobExecutor jobExecutor, final int parallelism) {
setAsContext(
- cluster,
+ jobExecutor,
parallelism,
- Collections.<Path>emptyList(),
- Collections.<URL>emptyList());
+ Collections.emptyList(),
+ Collections.emptyList());
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/057edf9e/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/TestEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/TestEnvironment.java b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/TestEnvironment.java
index ac5b17d..1d82f87 100644
--- a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/TestEnvironment.java
+++ b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/TestEnvironment.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.Plan;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.ExecutionEnvironmentFactory;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.optimizer.DataStatistics;
import org.apache.flink.optimizer.Optimizer;
@@ -30,6 +31,7 @@ import org.apache.flink.optimizer.plan.OptimizedPlan;
import org.apache.flink.optimizer.plandump.PlanJSONDumpGenerator;
import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.minicluster.JobExecutor;
import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster;
import org.apache.flink.util.Preconditions;
@@ -44,7 +46,7 @@ import java.util.Collections;
*/
public class TestEnvironment extends ExecutionEnvironment {
- private final LocalFlinkMiniCluster miniCluster;
+ private final JobExecutor jobExecutor;
private final Collection<Path> jarFiles;
@@ -53,12 +55,12 @@ public class TestEnvironment extends ExecutionEnvironment {
private TestEnvironment lastEnv;
public TestEnvironment(
- LocalFlinkMiniCluster miniCluster,
+ JobExecutor jobExecutor,
int parallelism,
boolean isObjectReuseEnabled,
Collection<Path> jarFiles,
Collection<URL> classPaths) {
- this.miniCluster = Preconditions.checkNotNull(miniCluster);
+ this.jobExecutor = Preconditions.checkNotNull(jobExecutor);
this.jarFiles = Preconditions.checkNotNull(jarFiles);
this.classPaths = Preconditions.checkNotNull(classPaths);
@@ -77,15 +79,15 @@ public class TestEnvironment extends ExecutionEnvironment {
}
public TestEnvironment(
- LocalFlinkMiniCluster executor,
+ JobExecutor executor,
int parallelism,
boolean isObjectReuseEnabled) {
this(
executor,
parallelism,
isObjectReuseEnabled,
- Collections.<Path>emptyList(),
- Collections.<URL>emptyList());
+ Collections.emptyList(),
+ Collections.emptyList());
}
@Override
@@ -115,7 +117,7 @@ public class TestEnvironment extends ExecutionEnvironment {
jobGraph.setClasspaths(new ArrayList<>(classPaths));
- this.lastJobExecutionResult = miniCluster.submitJobAndWait(jobGraph, false);
+ this.lastJobExecutionResult = jobExecutor.executeJobBlocking(jobGraph);
return this.lastJobExecutionResult;
}
@@ -130,7 +132,7 @@ public class TestEnvironment extends ExecutionEnvironment {
private OptimizedPlan compileProgram(String jobName) {
Plan p = createProgramPlan(jobName);
- Optimizer pc = new Optimizer(new DataStatistics(), this.miniCluster.configuration());
+ Optimizer pc = new Optimizer(new DataStatistics(), new Configuration());
return pc.compile(p);
}
@@ -138,7 +140,7 @@ public class TestEnvironment extends ExecutionEnvironment {
ExecutionEnvironmentFactory factory = new ExecutionEnvironmentFactory() {
@Override
public ExecutionEnvironment createExecutionEnvironment() {
- lastEnv = new TestEnvironment(miniCluster, getParallelism(), getConfig().isObjectReuseEnabled());
+ lastEnv = new TestEnvironment(jobExecutor, getParallelism(), getConfig().isObjectReuseEnabled());
return lastEnv;
}
};
@@ -153,13 +155,13 @@ public class TestEnvironment extends ExecutionEnvironment {
* environment executes the given jobs on a Flink mini cluster with the given default
* parallelism and the additional jar files and class paths.
*
- * @param miniCluster The mini cluster on which to execute the jobs
+ * @param jobExecutor The executor to run the jobs on
* @param parallelism The default parallelism
* @param jarFiles Additional jar files to execute the job with
* @param classPaths Additional class paths to execute the job with
*/
public static void setAsContext(
- final LocalFlinkMiniCluster miniCluster,
+ final JobExecutor jobExecutor,
final int parallelism,
final Collection<Path> jarFiles,
final Collection<URL> classPaths) {
@@ -168,7 +170,7 @@ public class TestEnvironment extends ExecutionEnvironment {
@Override
public ExecutionEnvironment createExecutionEnvironment() {
return new TestEnvironment(
- miniCluster,
+ jobExecutor,
parallelism,
false,
jarFiles,
@@ -185,15 +187,15 @@ public class TestEnvironment extends ExecutionEnvironment {
* environment executes the given jobs on a Flink mini cluster with the given default
* parallelism and the additional jar files and class paths.
*
- * @param miniCluster The mini cluster on which to execute the jobs
+ * @param jobExecutor The executor to run the jobs on
* @param parallelism The default parallelism
*/
- public static void setAsContext(final LocalFlinkMiniCluster miniCluster, final int parallelism) {
+ public static void setAsContext(final JobExecutor jobExecutor, final int parallelism) {
setAsContext(
- miniCluster,
+ jobExecutor,
parallelism,
- Collections.<Path>emptyList(),
- Collections.<URL>emptyList());
+ Collections.emptyList(),
+ Collections.emptyList());
}
public static void unsetAsContext() {