You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2019/12/17 10:41:05 UTC
[flink] 03/04: [FLINK-15201][client] Remove verifications in
ContextEnv for detached execution
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
commit ef618b72ad14c2f0c209b42d9e7d3421e2805953
Author: tison <wa...@gmail.com>
AuthorDate: Thu Dec 12 00:37:59 2019 +0800
[FLINK-15201][client] Remove verifications in ContextEnv for detached execution
---
.../flink/client/program/ContextEnvironment.java | 14 -------------
.../client/program/ContextEnvironmentFactory.java | 14 -------------
.../apache/flink/client/program/ClientTest.java | 24 ----------------------
.../core/execution/DetachedJobExecutionResult.java | 2 --
4 files changed, 54 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/ContextEnvironment.java b/flink-clients/src/main/java/org/apache/flink/client/program/ContextEnvironment.java
index 48f4b27..4c816d9 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/ContextEnvironment.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/ContextEnvironment.java
@@ -19,7 +19,6 @@
package org.apache.flink.client.program;
import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.InvalidProgramException;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.configuration.Configuration;
@@ -43,8 +42,6 @@ public class ContextEnvironment extends ExecutionEnvironment {
private static final Logger LOG = LoggerFactory.getLogger(ExecutionEnvironment.class);
- private boolean alreadyCalled;
-
ContextEnvironment(
final ExecutorServiceLoader executorServiceLoader,
final Configuration configuration,
@@ -55,14 +52,10 @@ public class ContextEnvironment extends ExecutionEnvironment {
if (parallelism > 0) {
setParallelism(parallelism);
}
-
- this.alreadyCalled = false;
}
@Override
public JobExecutionResult execute(String jobName) throws Exception {
- verifyExecuteIsCalledOnceWhenInDetachedMode();
-
JobClient jobClient = executeAsync(jobName);
JobExecutionResult jobExecutionResult;
@@ -101,13 +94,6 @@ public class ContextEnvironment extends ExecutionEnvironment {
return jobClient;
}
- private void verifyExecuteIsCalledOnceWhenInDetachedMode() {
- if (alreadyCalled && !getConfiguration().getBoolean(DeploymentOptions.ATTACHED)) {
- throw new InvalidProgramException(DetachedJobExecutionResult.DETACHED_MESSAGE + DetachedJobExecutionResult.EXECUTE_TWICE_MESSAGE);
- }
- alreadyCalled = true;
- }
-
@Override
public String toString() {
return "Context Environment (parallelism = " + (getParallelism() == ExecutionConfig.PARALLELISM_DEFAULT ? "default" : getParallelism()) + ")";
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/ContextEnvironmentFactory.java b/flink-clients/src/main/java/org/apache/flink/client/program/ContextEnvironmentFactory.java
index 15769e2..9a3c54a 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/ContextEnvironmentFactory.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/ContextEnvironmentFactory.java
@@ -18,11 +18,9 @@
package org.apache.flink.client.program;
-import org.apache.flink.api.common.InvalidProgramException;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.ExecutionEnvironmentFactory;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.DeploymentOptions;
import org.apache.flink.core.execution.ExecutorServiceLoader;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -40,8 +38,6 @@ public class ContextEnvironmentFactory implements ExecutionEnvironmentFactory {
private final ClassLoader userCodeClassLoader;
- private boolean alreadyCalled;
-
public ContextEnvironmentFactory(
final ExecutorServiceLoader executorServiceLoader,
final Configuration configuration,
@@ -49,23 +45,13 @@ public class ContextEnvironmentFactory implements ExecutionEnvironmentFactory {
this.executorServiceLoader = checkNotNull(executorServiceLoader);
this.configuration = checkNotNull(configuration);
this.userCodeClassLoader = checkNotNull(userCodeClassLoader);
-
- this.alreadyCalled = false;
}
@Override
public ExecutionEnvironment createExecutionEnvironment() {
- verifyCreateIsCalledOnceWhenInDetachedMode();
return new ContextEnvironment(
executorServiceLoader,
configuration,
userCodeClassLoader);
}
-
- private void verifyCreateIsCalledOnceWhenInDetachedMode() {
- if (!configuration.getBoolean(DeploymentOptions.ATTACHED) && alreadyCalled) {
- throw new InvalidProgramException("Multiple environments cannot be created in detached mode");
- }
- alreadyCalled = true;
- }
}
diff --git a/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java b/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
index dc140f8..0a5a722 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java
@@ -127,17 +127,6 @@ public class ClientTest extends TestLogger {
@Test
public void testDetachedMode() throws Exception{
final ClusterClient<?> clusterClient = new MiniClusterClient(new Configuration(), MINI_CLUSTER_RESOURCE.getMiniCluster());
- try {
- PackagedProgram prg = PackagedProgram.newBuilder().setEntryPointClassName(TestExecuteTwice.class.getName()).build();
- final Configuration configuration = fromPackagedProgram(prg, 1, true);
-
- ClientUtils.executeProgram(new TestExecutorServiceLoader(clusterClient, plan), configuration, prg);
- fail(FAIL_MESSAGE);
- } catch (ProgramInvocationException e) {
- assertEquals(
- DetachedJobExecutionResult.DETACHED_MESSAGE + DetachedJobExecutionResult.EXECUTE_TWICE_MESSAGE,
- e.getCause().getMessage());
- }
try {
PackagedProgram prg = PackagedProgram.newBuilder().setEntryPointClassName(TestEager.class.getName()).build();
@@ -295,19 +284,6 @@ public class ClientTest extends TestLogger {
}
/**
- * Test job that calls {@link ExecutionEnvironment#execute()} twice.
- */
- public static final class TestExecuteTwice {
-
- public static void main(String[] args) throws Exception {
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- env.fromElements(1, 2).output(new DiscardingOutputFormat<Integer>());
- env.execute();
- env.fromElements(1, 2).collect();
- }
- }
-
- /**
* Test job that uses an eager sink.
*/
public static final class TestEager {
diff --git a/flink-core/src/main/java/org/apache/flink/core/execution/DetachedJobExecutionResult.java b/flink-core/src/main/java/org/apache/flink/core/execution/DetachedJobExecutionResult.java
index d84c600..e4cfe18 100644
--- a/flink-core/src/main/java/org/apache/flink/core/execution/DetachedJobExecutionResult.java
+++ b/flink-core/src/main/java/org/apache/flink/core/execution/DetachedJobExecutionResult.java
@@ -34,8 +34,6 @@ public final class DetachedJobExecutionResult extends JobExecutionResult {
public static final String DETACHED_MESSAGE = "Job was submitted in detached mode. ";
- public static final String EXECUTE_TWICE_MESSAGE = "Only one call to execute is allowed. ";
-
public static final String EAGER_FUNCTION_MESSAGE = "Please make sure your program doesn't call " +
"an eager execution function [collect, print, printToErr, count]. ";