diff --git a/flink-clients/pom.xml b/flink-clients/pom.xml
index fa2c5e5266e56..8781d7b7eb3df 100644
--- a/flink-clients/pom.xml
+++ b/flink-clients/pom.xml
@@ -82,6 +82,12 @@ under the License.
 			<scope>test</scope>
 			<type>test-jar</type>
 		</dependency>
+
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+			<version>${project.version}</version>
+		</dependency>
 	</dependencies>
 
 	<!-- More information on this:
diff --git a/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java b/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
index 3c6515db9eaa2..f1b663ee74934 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java
@@ -21,9 +21,9 @@
 import org.apache.flink.api.common.JobExecutionResult;
 import org.apache.flink.client.program.ClusterClient;
 import org.apache.flink.client.program.ContextEnvironment;
-import org.apache.flink.client.program.ContextEnvironmentFactory;
 import org.apache.flink.client.program.PackagedProgram;
 import org.apache.flink.client.program.ProgramInvocationException;
+import org.apache.flink.client.program.StreamContextEnvironment;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.CoreOptions;
 import org.apache.flink.configuration.DeploymentOptions;
@@ -128,16 +128,21 @@ public static void executeProgram(
 
 			LOG.info("Starting program (detached: {})", !configuration.getBoolean(DeploymentOptions.ATTACHED));
 
-			ContextEnvironmentFactory factory = new ContextEnvironmentFactory(
-					executorServiceLoader,
-					configuration,
-					userCodeClassLoader);
-			ContextEnvironment.setAsContext(factory);
+			ContextEnvironment.setAsContext(
+				executorServiceLoader,
+				configuration,
+				userCodeClassLoader);
+
+			StreamContextEnvironment.setAsContext(
+				executorServiceLoader,
+				configuration,
+				userCodeClassLoader);
 
 			try {
 				program.invokeInteractiveModeForExecution();
 			} finally {
-				ContextEnvironment.unsetContext();
+				ContextEnvironment.unsetAsContext();
+				StreamContextEnvironment.unsetAsContext();
 			}
 		} finally {
 			Thread.currentThread().setContextClassLoader(contextClassLoader);
diff --git a/flink-clients/src/main/java/org/apache/flink/client/FlinkPipelineTranslationUtil.java b/flink-clients/src/main/java/org/apache/flink/client/FlinkPipelineTranslationUtil.java
index f88c69dd52a0c..fe0941c78a9a8 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/FlinkPipelineTranslationUtil.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/FlinkPipelineTranslationUtil.java
@@ -54,45 +54,19 @@ public static String translateToJSONExecutionPlan(Pipeline pipeline) {
 	}
 
 	private static FlinkPipelineTranslator getPipelineTranslator(Pipeline pipeline) {
-		PlanTranslator planToJobGraphTransmogrifier = new PlanTranslator();
+		PlanTranslator planTranslator = new PlanTranslator();
 
-		if (planToJobGraphTransmogrifier.canTranslate(pipeline)) {
-			return planToJobGraphTransmogrifier;
+		if (planTranslator.canTranslate(pipeline)) {
+			return planTranslator;
 		}
 
-		FlinkPipelineTranslator streamGraphTranslator = reflectStreamGraphTranslator();
+		StreamGraphTranslator streamGraphTranslator = new StreamGraphTranslator();
 
-		if (!streamGraphTranslator.canTranslate(pipeline)) {
-			throw new RuntimeException("Translator " + streamGraphTranslator + " cannot translate "
-					+ "the given pipeline " + pipeline + ".");
+		if (streamGraphTranslator.canTranslate(pipeline)) {
+			return streamGraphTranslator;
 		}
-		return streamGraphTranslator;
-	}
-
-	private static FlinkPipelineTranslator reflectStreamGraphTranslator() {
-		// Try our luck with StreamGraph translation. We have to load a StreamGraphTranslator
-		// via reflection because the dependencies of flink-streaming-java are inverted compared
-		// to flink-java. For flink-java does not depend on runtime, clients or optimizer and
-		// we have the translation code in clients/optimizer. On the other hand,
-		// flink-streaming-java depends on runtime and clients.
 
-		Class<?> streamGraphTranslatorClass;
-		try {
-			streamGraphTranslatorClass = Class.forName(
-					"org.apache.flink.streaming.api.graph.StreamGraphTranslator",
-					true,
-					FlinkPipelineTranslationUtil.class.getClassLoader());
-		} catch (ClassNotFoundException e) {
-			throw new RuntimeException("Could not load StreamGraphTranslator.", e);
-		}
-
-		FlinkPipelineTranslator streamGraphTranslator;
-		try {
-			streamGraphTranslator =
-					(FlinkPipelineTranslator) streamGraphTranslatorClass.newInstance();
-		} catch (InstantiationException | IllegalAccessException e) {
-			throw new RuntimeException("Could not instantiate StreamGraphTranslator.", e);
-		}
-		return streamGraphTranslator;
+		throw new RuntimeException("Translator " + streamGraphTranslator + " cannot translate "
+			+ "the given pipeline " + pipeline + ".");
 	}
 }
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphTranslator.java b/flink-clients/src/main/java/org/apache/flink/client/StreamGraphTranslator.java
similarity index 95%
rename from flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphTranslator.java
rename to flink-clients/src/main/java/org/apache/flink/client/StreamGraphTranslator.java
index 4a43adbb2ed02..affd6877bfbd6 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphTranslator.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/StreamGraphTranslator.java
@@ -17,12 +17,12 @@
  *
  */
 
-package org.apache.flink.streaming.api.graph;
+package org.apache.flink.client;
 
 import org.apache.flink.api.dag.Pipeline;
-import org.apache.flink.client.FlinkPipelineTranslator;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.streaming.api.graph.StreamGraph;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
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 824f4fb6ba1b5..5b7afb464d7bc 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
@@ -21,6 +21,7 @@
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.JobExecutionResult;
 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.DetachedJobExecutionResult;
@@ -41,7 +42,7 @@ public class ContextEnvironment extends ExecutionEnvironment {
 
 	private static final Logger LOG = LoggerFactory.getLogger(ExecutionEnvironment.class);
 
-	ContextEnvironment(
+	public ContextEnvironment(
 			final PipelineExecutorServiceLoader executorServiceLoader,
 			final Configuration configuration,
 			final ClassLoader userCodeClassLoader) {
@@ -95,11 +96,18 @@ public String toString() {
 
 	// --------------------------------------------------------------------------------------------
 
-	public static void setAsContext(ContextEnvironmentFactory factory) {
+	public static void setAsContext(
+			final PipelineExecutorServiceLoader executorServiceLoader,
+			final Configuration configuration,
+			final ClassLoader userCodeClassLoader) {
+		ExecutionEnvironmentFactory factory = () -> new ContextEnvironment(
+			executorServiceLoader,
+			configuration,
+			userCodeClassLoader);
 		initializeContextEnvironment(factory);
 	}
 
-	public static void unsetContext() {
+	public static void unsetAsContext() {
 		resetContextEnvironment();
 	}
 }
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
deleted file mode 100644
index 1620da0c3da67..0000000000000
--- a/flink-clients/src/main/java/org/apache/flink/client/program/ContextEnvironmentFactory.java
+++ /dev/null
@@ -1,57 +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.client.program;
-
-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.execution.PipelineExecutorServiceLoader;
-
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
-/**
- * The factory that instantiates the environment to be used when running jobs that are
- * submitted through a pre-configured client connection.
- * This happens for example when a job is submitted from the command line.
- */
-public class ContextEnvironmentFactory implements ExecutionEnvironmentFactory {
-
-	private final PipelineExecutorServiceLoader executorServiceLoader;
-
-	private final Configuration configuration;
-
-	private final ClassLoader userCodeClassLoader;
-
-	public ContextEnvironmentFactory(
-			final PipelineExecutorServiceLoader executorServiceLoader,
-			final Configuration configuration,
-			final ClassLoader userCodeClassLoader) {
-		this.executorServiceLoader = checkNotNull(executorServiceLoader);
-		this.configuration = checkNotNull(configuration);
-		this.userCodeClassLoader = checkNotNull(userCodeClassLoader);
-	}
-
-	@Override
-	public ExecutionEnvironment createExecutionEnvironment() {
-		return new ContextEnvironment(
-				executorServiceLoader,
-				configuration,
-				userCodeClassLoader);
-	}
-}
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/OptimizerPlanEnvironment.java b/flink-clients/src/main/java/org/apache/flink/client/program/OptimizerPlanEnvironment.java
index 3612bc025d9bc..87ff1a4f515b7 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/OptimizerPlanEnvironment.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/OptimizerPlanEnvironment.java
@@ -22,143 +22,39 @@
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.ExecutionEnvironmentFactory;
 import org.apache.flink.core.execution.JobClient;
-import org.apache.flink.util.Preconditions;
-
-import javax.annotation.Nullable;
-
-import java.io.ByteArrayOutputStream;
-import java.io.PrintStream;
 
 /**
- * An {@link ExecutionEnvironment} that never executes a job but only extracts the {@link
- * org.apache.flink.api.dag.Pipeline}.
+ * An {@link ExecutionEnvironment} that never executes a job but only extracts the {@link Pipeline}.
  */
 public class OptimizerPlanEnvironment extends ExecutionEnvironment {
 
 	private Pipeline pipeline;
 
-	// ------------------------------------------------------------------------
-	//  Execution Environment methods
-	// ------------------------------------------------------------------------
-
-	@Override
-	public JobClient executeAsync(String jobName) throws Exception {
-		this.pipeline = createProgramPlan();
-
-		// do not go on with anything now!
-		throw new ProgramAbortException();
+	public Pipeline getPipeline() {
+		return pipeline;
 	}
 
-	/**
-	 * Retrieves the JobGraph from a PackagedProgram.
-	 * @param prog The program to run
-	 * @param suppressOutput Whether to suppress stdout/stderr. Output is always printed on errors.
-	 * @return The Flink batch or streaming plan
-	 * @throws ProgramInvocationException in case of errors.
-	 */
-	public Pipeline getPipeline(PackagedProgram prog, boolean suppressOutput) throws ProgramInvocationException {
-
-		final PrintStream originalOut = System.out;
-		final PrintStream originalErr = System.err;
-		final ByteArrayOutputStream stdOutBuffer;
-		final ByteArrayOutputStream stdErrBuffer;
-
-		if (suppressOutput) {
-			// temporarily write syserr and sysout to a byte array.
-			stdOutBuffer = new ByteArrayOutputStream();
-			System.setOut(new PrintStream(stdOutBuffer));
-			stdErrBuffer = new ByteArrayOutputStream();
-			System.setErr(new PrintStream(stdErrBuffer));
-		} else {
-			stdOutBuffer = null;
-			stdErrBuffer = null;
-		}
-
-		setAsContext();
-		try {
-			prog.invokeInteractiveModeForExecution();
-		}
-		catch (ProgramInvocationException e) {
-			throw e;
-		}
-		catch (Throwable t) {
-			// the invocation gets aborted with the preview plan
-			if (pipeline != null) {
-				return pipeline;
-			} else {
-				throw generateException(prog, "The program caused an error: ", t, stdOutBuffer, stdErrBuffer);
-			}
-		}
-		finally {
-			unsetAsContext();
-			if (suppressOutput) {
-				System.setOut(originalOut);
-				System.setErr(originalErr);
-			}
+	public OptimizerPlanEnvironment(int parallelism) {
+		if (parallelism > 0) {
+			setParallelism(parallelism);
 		}
-
-		throw generateException(prog, "The program plan could not be fetched - the program aborted pre-maturely.", stdOutBuffer, stdErrBuffer);
 	}
 
-	// ------------------------------------------------------------------------
+	@Override
+	public JobClient executeAsync(String jobName) {
+		pipeline = createProgramPlan();
 
-	private void setAsContext() {
-		ExecutionEnvironmentFactory factory = new ExecutionEnvironmentFactory() {
+		// do not go on with anything now!
+		throw new ProgramAbortException();
+	}
 
-			@Override
-			public ExecutionEnvironment createExecutionEnvironment() {
-				return OptimizerPlanEnvironment.this;
-			}
-		};
+	public void setAsContext() {
+		ExecutionEnvironmentFactory factory = () -> this;
 		initializeContextEnvironment(factory);
 	}
 
-	private void unsetAsContext() {
+	public void unsetAsContext() {
 		resetContextEnvironment();
 	}
 
-	// ------------------------------------------------------------------------
-
-	public void setPipeline(Pipeline pipeline){
-		this.pipeline = pipeline;
-	}
-
-	private static ProgramInvocationException generateException(
-			PackagedProgram prog,
-			String msg,
-			@Nullable ByteArrayOutputStream stdout,
-			@Nullable ByteArrayOutputStream stderr) {
-		return generateException(prog, msg, null, stdout, stderr);
-	}
-
-	private static ProgramInvocationException generateException(
-			PackagedProgram prog,
-			String msg,
-			@Nullable Throwable cause,
-			@Nullable ByteArrayOutputStream stdoutBuffer,
-			@Nullable ByteArrayOutputStream stderrBuffer) {
-		Preconditions.checkState((stdoutBuffer != null) == (stderrBuffer != null),
-				"Stderr/Stdout should either both be set or both be null.");
-		String stdout = "";
-		String stderr = "";
-		if (stdoutBuffer != null) {
-			stdout = stdoutBuffer.toString();
-			stderr = stderrBuffer.toString();
-		}
-		return new ProgramInvocationException(
-			String.format("%s\n\nClasspath: %s\n\nSystem.out: %s\n\nSystem.err: %s",
-				msg,
-				prog.getJobJarAndDependencies(),
-				stdout.length() == 0 ? "(none)" : stdout,
-				stderr.length() == 0 ? "(none)" : stderr),
-			cause);
-	}
-
-	/**
-	 * A special exception used to abort programs when the caller is only interested in the
-	 * program plan, rather than in the full execution.
-	 */
-	public static final class ProgramAbortException extends Error {
-		private static final long serialVersionUID = 1L;
-	}
 }
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgramUtils.java b/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgramUtils.java
index a231b879c0430..5761e950ed125 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgramUtils.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgramUtils.java
@@ -27,10 +27,16 @@
 
 import javax.annotation.Nullable;
 
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
 /**
  * Utility class for {@link PackagedProgram} related operations.
  */
-public class PackagedProgramUtils {
+public enum PackagedProgramUtils {
+	;
 
 	private static final String PYTHON_DRIVER_CLASS_NAME = "org.apache.flink.client.python.PythonDriver";
 
@@ -40,10 +46,10 @@ public class PackagedProgramUtils {
 	 * Creates a {@link JobGraph} with a specified {@link JobID}
 	 * from the given {@link PackagedProgram}.
 	 *
-	 * @param packagedProgram to extract the JobGraph from
-	 * @param configuration to use for the optimizer and job graph generator
+	 * @param packagedProgram    to extract the JobGraph from
+	 * @param configuration      to use for the optimizer and job graph generator
 	 * @param defaultParallelism for the JobGraph
-	 * @param jobID the pre-generated job id
+	 * @param jobID              the pre-generated job id
 	 * @return JobGraph extracted from the PackagedProgram
 	 * @throws ProgramInvocationException if the JobGraph generation failed
 	 */
@@ -70,10 +76,10 @@ public static JobGraph createJobGraph(
 	 * Creates a {@link JobGraph} with a random {@link JobID}
 	 * from the given {@link PackagedProgram}.
 	 *
-	 * @param packagedProgram to extract the JobGraph from
-	 * @param configuration to use for the optimizer and job graph generator
+	 * @param packagedProgram    to extract the JobGraph from
+	 * @param configuration      to use for the optimizer and job graph generator
 	 * @param defaultParallelism for the JobGraph
-	 * @param suppressOutput Whether to suppress stdout/stderr during interactive JobGraph creation.
+	 * @param suppressOutput     Whether to suppress stdout/stderr during interactive JobGraph creation.
 	 * @return JobGraph extracted from the PackagedProgram
 	 * @throws ProgramInvocationException if the JobGraph generation failed
 	 */
@@ -86,22 +92,72 @@ public static JobGraph createJobGraph(
 	}
 
 	public static Pipeline getPipelineFromProgram(
-			PackagedProgram prog,
+			PackagedProgram program,
 			int parallelism,
 			boolean suppressOutput) throws CompilerException, ProgramInvocationException {
 		final ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
+
+		Thread.currentThread().setContextClassLoader(program.getUserCodeClassLoader());
+
+		final PrintStream originalOut = System.out;
+		final PrintStream originalErr = System.err;
+		final ByteArrayOutputStream stdOutBuffer;
+		final ByteArrayOutputStream stdErrBuffer;
+
+		if (suppressOutput) {
+			// temporarily write STDERR and STDOUT to a byte array.
+			stdOutBuffer = new ByteArrayOutputStream();
+			System.setOut(new PrintStream(stdOutBuffer));
+			stdErrBuffer = new ByteArrayOutputStream();
+			System.setErr(new PrintStream(stdErrBuffer));
+		} else {
+			stdOutBuffer = null;
+			stdErrBuffer = null;
+		}
+
+		// temporary hack to support the optimizer plan preview
+		OptimizerPlanEnvironment benv = new OptimizerPlanEnvironment(parallelism);
+		benv.setAsContext();
+		StreamPlanEnvironment senv = new StreamPlanEnvironment(parallelism);
+		senv.setAsContext();
+
 		try {
-			Thread.currentThread().setContextClassLoader(prog.getUserCodeClassLoader());
+			program.invokeInteractiveModeForExecution();
+		} catch (Throwable t) {
+			if (benv.getPipeline() != null) {
+				return benv.getPipeline();
+			}
 
-			// temporary hack to support the optimizer plan preview
-			OptimizerPlanEnvironment env = new OptimizerPlanEnvironment();
-			if (parallelism > 0) {
-				env.setParallelism(parallelism);
+			if (senv.getPipeline() != null) {
+				return senv.getPipeline();
 			}
-			return env.getPipeline(prog, suppressOutput);
+
+			if (t instanceof ProgramInvocationException) {
+				throw t;
+			}
+
+			throw generateException(
+				program,
+				"The program caused an error: ",
+				t,
+				stdOutBuffer,
+				stdErrBuffer);
 		} finally {
+			benv.unsetAsContext();
+			senv.unsetAsContext();
+			if (suppressOutput) {
+				System.setOut(originalOut);
+				System.setErr(originalErr);
+			}
 			Thread.currentThread().setContextClassLoader(contextClassLoader);
 		}
+
+		throw generateException(
+			program,
+			"The program plan could not be fetched - the program aborted pre-maturely.",
+			null,
+			stdOutBuffer,
+			stdErrBuffer);
 	}
 
 	public static Boolean isPython(String entryPointClassName) {
@@ -109,5 +165,25 @@ public static Boolean isPython(String entryPointClassName) {
 			(entryPointClassName.equals(PYTHON_DRIVER_CLASS_NAME) || entryPointClassName.equals(PYTHON_GATEWAY_CLASS_NAME));
 	}
 
-	private PackagedProgramUtils() {}
+	private static ProgramInvocationException generateException(
+			PackagedProgram program,
+			String msg,
+			@Nullable Throwable cause,
+			@Nullable ByteArrayOutputStream stdoutBuffer,
+			@Nullable ByteArrayOutputStream stderrBuffer) {
+		checkState(
+			(stdoutBuffer != null) == (stderrBuffer != null),
+			"Stderr/Stdout should either both be set or both be null.");
+
+		final String stdout = (stdoutBuffer != null) ? stdoutBuffer.toString() : "";
+		final String stderr = (stderrBuffer != null) ? stderrBuffer.toString() : "";
+
+		return new ProgramInvocationException(
+			String.format("%s\n\nClasspath: %s\n\nSystem.out: %s\n\nSystem.err: %s",
+				msg,
+				program.getJobJarAndDependencies(),
+				stdout.length() == 0 ? "(none)" : stdout,
+				stderr.length() == 0 ? "(none)" : stderr),
+			cause);
+	}
 }
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/ProgramAbortException.java b/flink-clients/src/main/java/org/apache/flink/client/program/ProgramAbortException.java
new file mode 100644
index 0000000000000..a77e43a29fa88
--- /dev/null
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/ProgramAbortException.java
@@ -0,0 +1,27 @@
+/*
+ * 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.client.program;
+
+/**
+ * A special exception used to abort programs when the caller is only interested in the
+ * program plan, rather than in the full execution.
+ */
+public class ProgramAbortException extends Error {
+	private static final long serialVersionUID = 1L;
+}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamContextEnvironment.java b/flink-clients/src/main/java/org/apache/flink/client/program/StreamContextEnvironment.java
similarity index 69%
rename from flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamContextEnvironment.java
rename to flink-clients/src/main/java/org/apache/flink/client/program/StreamContextEnvironment.java
index b4ecbbfd7d932..1778202cd90da 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamContextEnvironment.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/StreamContextEnvironment.java
@@ -15,15 +15,18 @@
  * limitations under the License.
  */
 
-package org.apache.flink.streaming.api.environment;
+package org.apache.flink.client.program;
 
 import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.api.common.JobExecutionResult;
 import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.client.program.ContextEnvironment;
+import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.DeploymentOptions;
 import org.apache.flink.core.execution.DetachedJobExecutionResult;
 import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.PipelineExecutorServiceLoader;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironmentFactory;
 import org.apache.flink.streaming.api.graph.StreamGraph;
 import org.apache.flink.util.ShutdownHookUtil;
 
@@ -33,8 +36,6 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
 /**
  * Special {@link StreamExecutionEnvironment} that will be used in cases where the CLI client or
  * testing utilities create a {@link StreamExecutionEnvironment} that should be used when
@@ -45,17 +46,11 @@ public class StreamContextEnvironment extends StreamExecutionEnvironment {
 
 	private static final Logger LOG = LoggerFactory.getLogger(ExecutionEnvironment.class);
 
-	private final ContextEnvironment ctx;
-
-	StreamContextEnvironment(final ContextEnvironment ctx) {
-		super(checkNotNull(ctx).getExecutorServiceLoader(), ctx.getConfiguration(), ctx.getUserCodeClassLoader());
-
-		this.ctx = ctx;
-
-		final int parallelism = ctx.getParallelism();
-		if (parallelism > 0) {
-			setParallelism(parallelism);
-		}
+	public StreamContextEnvironment(
+			final PipelineExecutorServiceLoader executorServiceLoader,
+			final Configuration configuration,
+			final ClassLoader userCodeClassLoader) {
+		super(executorServiceLoader, configuration, userCodeClassLoader);
 	}
 
 	@Override
@@ -65,7 +60,7 @@ public JobExecutionResult execute(StreamGraph streamGraph) throws Exception {
 		JobExecutionResult jobExecutionResult;
 		if (getConfiguration().getBoolean(DeploymentOptions.ATTACHED)) {
 			CompletableFuture<JobExecutionResult> jobExecutionResultFuture =
-					jobClient.getJobExecutionResult(ctx.getUserCodeClassLoader());
+					jobClient.getJobExecutionResult(getUserClassloader());
 
 			if (getConfiguration().getBoolean(DeploymentOptions.SHUTDOWN_IF_ATTACHED)) {
 				Thread shutdownHook = ShutdownHookUtil.addShutdownHook(
@@ -74,10 +69,11 @@ public JobExecutionResult execute(StreamGraph streamGraph) throws Exception {
 							// the jvm exits
 							jobClient.cancel().get(1, TimeUnit.SECONDS);
 						},
-						ContextEnvironment.class.getSimpleName(),
+						StreamContextEnvironment.class.getSimpleName(),
 						LOG);
 				jobExecutionResultFuture.whenComplete((ignored, throwable) ->
-						ShutdownHookUtil.removeShutdownHook(shutdownHook, ContextEnvironment.class.getSimpleName(), LOG));
+						ShutdownHookUtil.removeShutdownHook(
+							shutdownHook, StreamContextEnvironment.class.getSimpleName(), LOG));
 			}
 
 			jobExecutionResult = jobExecutionResultFuture.get();
@@ -97,4 +93,21 @@ public JobClient executeAsync(StreamGraph streamGraph) throws Exception {
 
 		return jobClient;
 	}
+
+	// --------------------------------------------------------------------------------------------
+
+	public static void setAsContext(
+			final PipelineExecutorServiceLoader executorServiceLoader,
+			final Configuration configuration,
+			final ClassLoader userCodeClassLoader) {
+		StreamExecutionEnvironmentFactory factory = () -> new StreamContextEnvironment(
+			executorServiceLoader,
+			configuration,
+			userCodeClassLoader);
+		initializeContextEnvironment(factory);
+	}
+
+	public static void unsetAsContext() {
+		resetContextEnvironment();
+	}
 }
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamPlanEnvironment.java b/flink-clients/src/main/java/org/apache/flink/client/program/StreamPlanEnvironment.java
similarity index 60%
rename from flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamPlanEnvironment.java
rename to flink-clients/src/main/java/org/apache/flink/client/program/StreamPlanEnvironment.java
index eafeb9e7628ab..86b05af90663a 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamPlanEnvironment.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/StreamPlanEnvironment.java
@@ -15,14 +15,13 @@
  * limitations under the License.
  */
 
-package org.apache.flink.streaming.api.environment;
+package org.apache.flink.client.program;
 
 import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.client.program.OptimizerPlanEnvironment;
-import org.apache.flink.configuration.CoreOptions;
-import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.api.dag.Pipeline;
 import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironmentFactory;
 import org.apache.flink.streaming.api.graph.StreamGraph;
 
 /**
@@ -32,26 +31,32 @@
 @PublicEvolving
 public class StreamPlanEnvironment extends StreamExecutionEnvironment {
 
-	private ExecutionEnvironment env;
+	private Pipeline pipeline;
 
-	protected StreamPlanEnvironment(ExecutionEnvironment env) {
-		this.env = env;
+	public Pipeline getPipeline() {
+		return pipeline;
+	}
 
-		int parallelism = env.getParallelism();
+	public StreamPlanEnvironment(int parallelism) {
 		if (parallelism > 0) {
 			setParallelism(parallelism);
-		} else {
-			// determine parallelism
-			setParallelism(GlobalConfiguration.loadConfiguration().getInteger(CoreOptions.DEFAULT_PARALLELISM));
 		}
 	}
 
 	@Override
-	public JobClient executeAsync(StreamGraph streamGraph) throws Exception {
-		if (env instanceof OptimizerPlanEnvironment) {
-			((OptimizerPlanEnvironment) env).setPipeline(streamGraph);
-		}
+	public JobClient executeAsync(StreamGraph streamGraph) {
+		pipeline = streamGraph;
+
+		// do not go on with anything now!
+		throw new ProgramAbortException();
+	}
+
+	public void setAsContext() {
+		StreamExecutionEnvironmentFactory factory = () -> this;
+		initializeContextEnvironment(factory);
+	}
 
-		throw new OptimizerPlanEnvironment.ProgramAbortException();
+	public void unsetAsContext() {
+		resetContextEnvironment();
 	}
 }
diff --git a/flink-clients/src/test/java/org/apache/flink/client/program/OptimizerPlanEnvironmentTest.java b/flink-clients/src/test/java/org/apache/flink/client/program/OptimizerPlanEnvironmentTest.java
index 96a8a7c51adcd..0bd89e194c9a1 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/program/OptimizerPlanEnvironmentTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/program/OptimizerPlanEnvironmentTest.java
@@ -44,10 +44,10 @@ private void runOutputTest(boolean suppressOutput, String[] expectedCapturedOutp
 		PackagedProgram packagedProgram = PackagedProgram.newBuilder()
 			.setEntryPointClassName(getClass().getName())
 			.build();
-		OptimizerPlanEnvironment env = new OptimizerPlanEnvironment();
+
 		try {
 			// Flink will throw an error because no job graph will be generated by the main method.
-			env.getPipeline(packagedProgram, suppressOutput);
+			PackagedProgramUtils.getPipelineFromProgram(packagedProgram, 1, suppressOutput);
 			Assert.fail("This should have failed to create the Flink Plan.");
 		} catch (ProgramInvocationException e) {
 			// Test that that Flink captured the expected stdout/stderr
diff --git a/flink-python/pom.xml b/flink-python/pom.xml
index 7f2938391bd18..fc4b97ad3e612 100644
--- a/flink-python/pom.xml
+++ b/flink-python/pom.xml
@@ -44,6 +44,12 @@ under the License.
 			<version>${project.version}</version>
 			<scope>provided</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-clients_${scala.binary.version}</artifactId>
+			<version>${project.version}</version>
+			<scope>provided</scope>
+		</dependency>
 		<dependency>
 			<groupId>org.apache.flink</groupId>
 			<artifactId>flink-java</artifactId>
diff --git a/flink-python/src/main/java/org/apache/flink/client/python/PythonDriver.java b/flink-python/src/main/java/org/apache/flink/client/python/PythonDriver.java
index 6fcb190b5a46c..bd6143a9b61be 100644
--- a/flink-python/src/main/java/org/apache/flink/client/python/PythonDriver.java
+++ b/flink-python/src/main/java/org/apache/flink/client/python/PythonDriver.java
@@ -18,7 +18,7 @@
 
 package org.apache.flink.client.python;
 
-import org.apache.flink.client.program.OptimizerPlanEnvironment;
+import org.apache.flink.client.program.ProgramAbortException;
 import org.apache.flink.runtime.entrypoint.parser.CommandLineParser;
 
 import org.slf4j.Logger;
@@ -84,7 +84,7 @@ public static void main(String[] args) {
 
 			// throw ProgramAbortException if the caller is interested in the program plan,
 			// there is no harm to throw ProgramAbortException even if it is not the case.
-			throw new OptimizerPlanEnvironment.ProgramAbortException();
+			throw new ProgramAbortException();
 		} finally {
 			gatewayServer.shutdown();
 		}
diff --git a/flink-streaming-java/pom.xml b/flink-streaming-java/pom.xml
index 9719cfa78a36f..41f3381e0264f 100644
--- a/flink-streaming-java/pom.xml
+++ b/flink-streaming-java/pom.xml
@@ -52,7 +52,7 @@ under the License.
 
 		<dependency>
 			<groupId>org.apache.flink</groupId>
-			<artifactId>flink-clients_${scala.binary.version}</artifactId>
+			<artifactId>flink-java</artifactId>
 			<version>${project.version}</version>
 		</dependency>
 
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
index cfc45a72ed2bb..7a852eed9219c 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
@@ -34,7 +34,6 @@
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.dag.Transformation;
 import org.apache.flink.api.java.ClosureCleaner;
-import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.Utils;
 import org.apache.flink.api.java.io.TextInputFormat;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -43,8 +42,6 @@
 import org.apache.flink.api.java.typeutils.PojoTypeInfo;
 import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
 import org.apache.flink.api.java.typeutils.TypeExtractor;
-import org.apache.flink.client.program.ContextEnvironment;
-import org.apache.flink.client.program.OptimizerPlanEnvironment;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.DeploymentOptions;
 import org.apache.flink.configuration.ExecutionOptions;
@@ -222,6 +219,10 @@ protected Configuration getConfiguration() {
 		return this.configuration;
 	}
 
+	protected ClassLoader getUserClassloader() {
+		return userClassloader;
+	}
+
 	/**
 	 * Gets the config object.
 	 */
@@ -1855,22 +1856,7 @@ public void addOperator(Transformation<?> transformation) {
 	public static StreamExecutionEnvironment getExecutionEnvironment() {
 		return Utils.resolveFactory(threadLocalContextEnvironmentFactory, contextEnvironmentFactory)
 			.map(StreamExecutionEnvironmentFactory::createExecutionEnvironment)
-			.orElseGet(StreamExecutionEnvironment::createStreamExecutionEnvironment);
-	}
-
-	private static StreamExecutionEnvironment createStreamExecutionEnvironment() {
-		// because the streaming project depends on "flink-clients" (and not the other way around)
-		// we currently need to intercept the data set environment and create a dependent stream env.
-		// this should be fixed once we rework the project dependencies
-
-		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-		if (env instanceof ContextEnvironment) {
-			return new StreamContextEnvironment((ContextEnvironment) env);
-		} else if (env instanceof OptimizerPlanEnvironment) {
-			return new StreamPlanEnvironment(env);
-		} else {
-			return createLocalEnvironment();
-		}
+			.orElseGet(StreamExecutionEnvironment::createLocalEnvironment);
 	}
 
 	/**
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/graph/StreamingJobGraphGeneratorNodeHashTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/graph/StreamingJobGraphGeneratorNodeHashTest.java
index 0726261c98f7f..d00d0af265c25 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/graph/StreamingJobGraphGeneratorNodeHashTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/graph/StreamingJobGraphGeneratorNodeHashTest.java
@@ -491,7 +491,7 @@ public void testDisablingAutoUidsAcceptsManuallySetHash() {
 
 	@Test
 	public void testDisablingAutoUidsWorksWithKeyBy() throws Exception {
-		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+		StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
 		env.getConfig().disableAutoGeneratedUIDs();
 
 		env
@@ -499,7 +499,7 @@ public void testDisablingAutoUidsWorksWithKeyBy() throws Exception {
 			.keyBy(o -> o)
 			.addSink(new DiscardingSink<>()).setUidHash("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb");
 
-		env.execute();
+		env.getStreamGraph();
 	}
 
 	// ------------------------------------------------------------------------