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 2020/01/11 13:36:14 UTC
[flink] branch release-1.10 updated: [FLINK-15504] Allow output to
stdout/stderr during execution of PackagedProgram
This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.10 by this push:
new 0ea4dd7 [FLINK-15504] Allow output to stdout/stderr during execution of PackagedProgram
0ea4dd7 is described below
commit 0ea4dd7e9d56a017743ca6794d28537800faab6f
Author: Maximilian Michels <mx...@apache.org>
AuthorDate: Thu Jan 9 12:23:03 2020 +0100
[FLINK-15504] Allow output to stdout/stderr during execution of PackagedProgram
We suppress the output to stdout/stderr during plan extraction via
PackagedProgram. This has unintended consequences for users who are looking into
debugging their Flink programs during JobGraph creation.
This change removes the suppression of output when we run the JARs. The plan
preview still suppresses the output to avoid spaming the logs during plan
preview.
---
.../org/apache/flink/client/cli/CliFrontend.java | 2 +-
.../client/program/OptimizerPlanEnvironment.java | 84 +++++++---
.../flink/client/program/PackagedProgramUtils.java | 17 ++-
.../client/cli/CliFrontendPackageProgramTest.java | 2 +-
.../apache/flink/client/program/ClientTest.java | 2 +-
.../client/program/ExecutionPlanCreationTest.java | 2 +-
.../program/OptimizerPlanEnvironmentTest.java | 70 +++++++++
.../entrypoint/ClassPathJobGraphRetriever.java | 3 +-
.../webmonitor/handlers/JarPlanHandler.java | 2 +-
.../runtime/webmonitor/handlers/JarRunHandler.java | 2 +-
.../webmonitor/handlers/utils/JarHandlerUtils.java | 4 +-
.../webmonitor/handlers/JarHandlerTest.java | 170 +++++++++++++++++++++
.../webmonitor/handlers/JarRunHandlerTest.java | 114 --------------
.../optimizer/jsonplan/DumpCompiledPlanTest.java | 2 +-
.../optimizer/jsonplan/PreviewPlanDumpTest.java | 2 +-
.../apache/flink/yarn/YarnConfigurationITCase.java | 2 +-
16 files changed, 329 insertions(+), 151 deletions(-)
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
index e834c83..c568f63 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java
@@ -268,7 +268,7 @@ public class CliFrontend {
LOG.info("Creating program plan dump");
- Pipeline pipeline = PackagedProgramUtils.getPipelineFromProgram(program, parallelism);
+ Pipeline pipeline = PackagedProgramUtils.getPipelineFromProgram(program, parallelism, true);
String jsonPlan = FlinkPipelineTranslationUtil.translateToJSONExecutionPlan(pipeline);
if (jsonPlan != null) {
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 d997965..3612bc0 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,6 +22,9 @@ import org.apache.flink.api.dag.Pipeline;
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;
@@ -46,15 +49,30 @@ public class OptimizerPlanEnvironment extends ExecutionEnvironment {
throw new ProgramAbortException();
}
- public Pipeline getPipeline(PackagedProgram prog) throws ProgramInvocationException {
-
- // temporarily write syserr and sysout to a byte array.
- PrintStream originalOut = System.out;
- PrintStream originalErr = System.err;
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- System.setOut(new PrintStream(baos));
- ByteArrayOutputStream baes = new ByteArrayOutputStream();
- System.setErr(new PrintStream(baes));
+ /**
+ * 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 {
@@ -68,23 +86,20 @@ public class OptimizerPlanEnvironment extends ExecutionEnvironment {
if (pipeline != null) {
return pipeline;
} else {
- throw new ProgramInvocationException("The program caused an error: ", t);
+ throw generateException(prog, "The program caused an error: ", t, stdOutBuffer, stdErrBuffer);
}
}
finally {
unsetAsContext();
- System.setOut(originalOut);
- System.setErr(originalErr);
+ if (suppressOutput) {
+ System.setOut(originalOut);
+ System.setErr(originalErr);
+ }
}
- String stdout = baos.toString();
- String stderr = baes.toString();
-
- throw new ProgramInvocationException(
- "The program plan could not be fetched - the program aborted pre-maturely."
- + "\n\nSystem.err: " + (stderr.length() == 0 ? "(none)" : stderr)
- + "\n\nSystem.out: " + (stdout.length() == 0 ? "(none)" : stdout));
+ throw generateException(prog, "The program plan could not be fetched - the program aborted pre-maturely.", stdOutBuffer, stdErrBuffer);
}
+
// ------------------------------------------------------------------------
private void setAsContext() {
@@ -108,6 +123,37 @@ public class OptimizerPlanEnvironment extends ExecutionEnvironment {
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.
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 0751166..a231b87 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
@@ -35,6 +35,7 @@ public class PackagedProgramUtils {
private static final String PYTHON_DRIVER_CLASS_NAME = "org.apache.flink.client.python.PythonDriver";
private static final String PYTHON_GATEWAY_CLASS_NAME = "org.apache.flink.client.python.PythonGatewayServer";
+
/**
* Creates a {@link JobGraph} with a specified {@link JobID}
* from the given {@link PackagedProgram}.
@@ -50,8 +51,9 @@ public class PackagedProgramUtils {
PackagedProgram packagedProgram,
Configuration configuration,
int defaultParallelism,
- @Nullable JobID jobID) throws ProgramInvocationException {
- final Pipeline pipeline = getPipelineFromProgram(packagedProgram, defaultParallelism);
+ @Nullable JobID jobID,
+ boolean suppressOutput) throws ProgramInvocationException {
+ final Pipeline pipeline = getPipelineFromProgram(packagedProgram, defaultParallelism, suppressOutput);
final JobGraph jobGraph = FlinkPipelineTranslationUtil.getJobGraph(pipeline, configuration, defaultParallelism);
if (jobID != null) {
@@ -71,19 +73,22 @@ public class PackagedProgramUtils {
* @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.
* @return JobGraph extracted from the PackagedProgram
* @throws ProgramInvocationException if the JobGraph generation failed
*/
public static JobGraph createJobGraph(
PackagedProgram packagedProgram,
Configuration configuration,
- int defaultParallelism) throws ProgramInvocationException {
- return createJobGraph(packagedProgram, configuration, defaultParallelism, null);
+ int defaultParallelism,
+ boolean suppressOutput) throws ProgramInvocationException {
+ return createJobGraph(packagedProgram, configuration, defaultParallelism, null, suppressOutput);
}
public static Pipeline getPipelineFromProgram(
PackagedProgram prog,
- int parallelism) throws CompilerException, ProgramInvocationException {
+ int parallelism,
+ boolean suppressOutput) throws CompilerException, ProgramInvocationException {
final ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
try {
Thread.currentThread().setContextClassLoader(prog.getUserCodeClassLoader());
@@ -93,7 +98,7 @@ public class PackagedProgramUtils {
if (parallelism > 0) {
env.setParallelism(parallelism);
}
- return env.getPipeline(prog);
+ return env.getPipeline(prog, suppressOutput);
} finally {
Thread.currentThread().setContextClassLoader(contextClassLoader);
}
diff --git a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendPackageProgramTest.java b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendPackageProgramTest.java
index 8261a57..dd8f345 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendPackageProgramTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/cli/CliFrontendPackageProgramTest.java
@@ -301,7 +301,7 @@ public class CliFrontendPackageProgramTest extends TestLogger {
Optimizer compiler = new Optimizer(new DataStatistics(), new DefaultCostEstimator(), c);
// we expect this to fail with a "ClassNotFoundException"
- Pipeline pipeline = PackagedProgramUtils.getPipelineFromProgram(prog, 666);
+ Pipeline pipeline = PackagedProgramUtils.getPipelineFromProgram(prog, 666, true);
FlinkPipelineTranslationUtil.translateToJSONExecutionPlan(pipeline);
fail("Should have failed with a ClassNotFoundException");
}
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 0a5a722..c5a88b2 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
@@ -233,7 +233,7 @@ public class ClientTest extends TestLogger {
.build();
Optimizer optimizer = new Optimizer(new DataStatistics(), new DefaultCostEstimator(), config);
- Plan plan = (Plan) PackagedProgramUtils.getPipelineFromProgram(prg, 1);
+ Plan plan = (Plan) PackagedProgramUtils.getPipelineFromProgram(prg, 1, true);
OptimizedPlan op = optimizer.compile(plan);
assertNotNull(op);
diff --git a/flink-clients/src/test/java/org/apache/flink/client/program/ExecutionPlanCreationTest.java b/flink-clients/src/test/java/org/apache/flink/client/program/ExecutionPlanCreationTest.java
index 6343a2f..0c258e4 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/program/ExecutionPlanCreationTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/program/ExecutionPlanCreationTest.java
@@ -63,7 +63,7 @@ public class ExecutionPlanCreationTest {
config.setInteger(JobManagerOptions.PORT, mockJmAddress.getPort());
Optimizer optimizer = new Optimizer(new DataStatistics(), new DefaultCostEstimator(), config);
- Plan plan = (Plan) PackagedProgramUtils.getPipelineFromProgram(prg, -1);
+ Plan plan = (Plan) PackagedProgramUtils.getPipelineFromProgram(prg, -1, true);
OptimizedPlan op = optimizer.compile(plan);
assertNotNull(op);
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
new file mode 100644
index 0000000..96a8a7c
--- /dev/null
+++ b/flink-clients/src/test/java/org/apache/flink/client/program/OptimizerPlanEnvironmentTest.java
@@ -0,0 +1,70 @@
+/*
+ * 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.junit.Assert;
+import org.junit.Test;
+
+import static org.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+/**
+ * Tests for {@link OptimizerPlanEnvironment}.
+ */
+public class OptimizerPlanEnvironmentTest {
+
+ /**
+ * Test the two modes for handling stdout/stderr of user program.
+ * (1) Capturing the output and including it only in the exception
+ * (2) Leaving the output untouched
+ */
+ @Test
+ public void testStdOutStdErrHandling() throws Exception {
+ runOutputTest(true, new String[] {"System.out: hello out!", "System.err: hello err!"});
+ runOutputTest(false, new String[] {"System.out: (none)", "System.err: (none)"});
+ }
+
+ private void runOutputTest(boolean suppressOutput, String[] expectedCapturedOutput) throws ProgramInvocationException {
+ 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);
+ Assert.fail("This should have failed to create the Flink Plan.");
+ } catch (ProgramInvocationException e) {
+ // Test that that Flink captured the expected stdout/stderr
+ for (String expected : expectedCapturedOutput) {
+ assertThat(e.getMessage(), containsString(expected));
+ }
+ }
+ }
+
+ /**
+ * Main method for {@code testEnsureStdoutStdErrIsRestored()}.
+ * This will not create a valid Flink program. We will just use this program to check whether stdout/stderr is
+ * captured in a byte buffer or directly printed to the console.
+ * */
+ public static void main(String[] args) {
+ // Print something to stdout/stderr for output suppression test
+ System.out.println("hello out!");
+ System.err.println("hello err!");
+ }
+}
diff --git a/flink-container/src/main/java/org/apache/flink/container/entrypoint/ClassPathJobGraphRetriever.java b/flink-container/src/main/java/org/apache/flink/container/entrypoint/ClassPathJobGraphRetriever.java
index 9ed109a..42b5cb1 100644
--- a/flink-container/src/main/java/org/apache/flink/container/entrypoint/ClassPathJobGraphRetriever.java
+++ b/flink-container/src/main/java/org/apache/flink/container/entrypoint/ClassPathJobGraphRetriever.java
@@ -105,7 +105,8 @@ class ClassPathJobGraphRetriever extends AbstractUserClassPathJobGraphRetriever
packagedProgram,
configuration,
defaultParallelism,
- jobId);
+ jobId,
+ false);
jobGraph.setSavepointRestoreSettings(savepointRestoreSettings);
return jobGraph;
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarPlanHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarPlanHandler.java
index 6fb84d4..2327196 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarPlanHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarPlanHandler.java
@@ -97,7 +97,7 @@ public class JarPlanHandler
final JarHandlerContext context = JarHandlerContext.fromRequest(request, jarDir, log);
return CompletableFuture.supplyAsync(() -> {
- final JobGraph jobGraph = context.toJobGraph(configuration);
+ final JobGraph jobGraph = context.toJobGraph(configuration, true);
return planGenerator.apply(jobGraph);
}, executor);
}
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarRunHandler.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarRunHandler.java
index dcce5a9..d5b7248 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarRunHandler.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarRunHandler.java
@@ -135,7 +135,7 @@ public class JarRunHandler extends
JarHandlerContext context,
final SavepointRestoreSettings savepointRestoreSettings) {
return CompletableFuture.supplyAsync(() -> {
- final JobGraph jobGraph = context.toJobGraph(configuration);
+ final JobGraph jobGraph = context.toJobGraph(configuration, false);
jobGraph.setSavepointRestoreSettings(savepointRestoreSettings);
return jobGraph;
}, executor);
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/utils/JarHandlerUtils.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/utils/JarHandlerUtils.java
index 9ceb1ae..4b0da5c 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/utils/JarHandlerUtils.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/utils/JarHandlerUtils.java
@@ -112,7 +112,7 @@ public class JarHandlerUtils {
return new JarHandlerContext(jarFile, entryClass, programArgs, parallelism, jobId);
}
- public JobGraph toJobGraph(Configuration configuration) {
+ public JobGraph toJobGraph(Configuration configuration, boolean suppressOutput) {
if (!Files.exists(jarFile)) {
throw new CompletionException(new RestHandlerException(
String.format("Jar file %s does not exist", jarFile), HttpResponseStatus.BAD_REQUEST));
@@ -125,7 +125,7 @@ public class JarHandlerUtils {
.setConfiguration(configuration)
.setArguments(programArgs.toArray(new String[0]))
.build();
- return PackagedProgramUtils.createJobGraph(packagedProgram, configuration, parallelism, jobId);
+ return PackagedProgramUtils.createJobGraph(packagedProgram, configuration, parallelism, jobId, suppressOutput);
} catch (final ProgramInvocationException e) {
throw new CompletionException(e);
}
diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/handlers/JarHandlerTest.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/handlers/JarHandlerTest.java
new file mode 100644
index 0000000..c565322
--- /dev/null
+++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/handlers/JarHandlerTest.java
@@ -0,0 +1,170 @@
+/*
+ * 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.webmonitor.handlers;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.configuration.WebOptions;
+import org.apache.flink.runtime.rest.RestClient;
+import org.apache.flink.runtime.rest.RestClientConfiguration;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.runtime.rest.util.RestClientException;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
+import org.apache.flink.runtime.testutils.MiniClusterResource;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.testutils.junit.category.AlsoRunWithLegacyScheduler;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.FileSystem;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Optional;
+
+import static org.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+/**
+ * Tests for the {@link JarRunHandler} and {@link JarPlanHandler}.
+ */
+@Category(AlsoRunWithLegacyScheduler.class)
+public class JarHandlerTest extends TestLogger {
+
+ @ClassRule
+ public static final TemporaryFolder TMP = new TemporaryFolder();
+
+ enum Type {
+ PLAN,
+ RUN
+ }
+
+ @Test
+ public void testPlanJar() throws Exception {
+ runTest(Type.PLAN, "hello out!", "hello err!");
+ }
+
+ @Test
+ public void testRunJar() throws Exception {
+ runTest(Type.RUN, "(none)", "(none)");
+ }
+
+ private static void runTest(Type type, String expectedCapturedStdOut, String expectedCapturedStdErr) throws Exception {
+ Path uploadDir = TMP.newFolder().toPath();
+
+ Path actualUploadDir = uploadDir.resolve("flink-web-upload");
+ Files.createDirectory(actualUploadDir);
+
+ Path emptyJar = actualUploadDir.resolve("empty.jar");
+ createJarFile(emptyJar);
+
+ Configuration config = new Configuration();
+ config.setString(WebOptions.UPLOAD_DIR, uploadDir.toString());
+
+ MiniClusterResource clusterResource = new MiniClusterResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setConfiguration(config)
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(1)
+ .build());
+ clusterResource.before();
+
+ try {
+ Configuration clientConfig = clusterResource.getClientConfiguration();
+ RestClient client = new RestClient(RestClientConfiguration.fromConfiguration(clientConfig), TestingUtils.defaultExecutor());
+
+ try {
+ final MessageHeaders headers;
+ final JarMessageParameters parameters;
+ if (type == Type.RUN) {
+ headers = JarRunHeaders.getInstance();
+ parameters = ((JarRunHeaders) headers).getUnresolvedMessageParameters();
+ } else if (type == Type.PLAN) {
+ headers = JarPlanGetHeaders.getInstance();
+ parameters = ((JarPlanGetHeaders) headers).getUnresolvedMessageParameters();
+ } else {
+ throw new RuntimeException("Invalid type: " + type);
+ }
+ parameters.jarIdPathParameter.resolve(emptyJar.getFileName().toString());
+
+ String host = clientConfig.getString(RestOptions.ADDRESS);
+ int port = clientConfig.getInteger(RestOptions.PORT);
+
+ try {
+ client.sendRequest(host, port, headers, parameters, new JarPlanRequestBody()).get();
+ } catch (Exception e) {
+ Optional<RestClientException> expected = ExceptionUtils.findThrowable(e, RestClientException.class);
+ if (expected.isPresent()) {
+ String message = expected.get().getMessage();
+ // implies the job was actually submitted
+ assertThat(message, containsString("ProgramInvocationException"));
+ // original cause is preserved in stack trace
+ assertThat(message, containsString("The program plan could not be fetched - the program aborted pre-maturely"));
+ // implies the jar was registered for the job graph (otherwise the jar name would not occur in the exception)
+ // implies the jar was uploaded (otherwise the file would not be found at all)
+ assertThat(message, containsString("empty.jar"));
+ // ensure that no stdout/stderr has been captured
+ assertThat(message, containsString("System.out: " + expectedCapturedStdOut));
+ assertThat(message, containsString("System.err: " + expectedCapturedStdErr));
+ } else {
+ throw e;
+ }
+ }
+ } finally {
+ client.shutdown(Time.milliseconds(10));
+ }
+ } finally {
+ clusterResource.after();
+ }
+ }
+
+ private static void createJarFile(Path zipFile) throws IOException {
+ URI uri = URI.create("jar:file:" + zipFile.toString());
+ HashMap<String, Object> env = new HashMap<>();
+ // We need this to ensure the file will be created if it does not exist
+ env.put("create", "true");
+ try (FileSystem zipfs = FileSystems.newFileSystem(uri, env)) {
+ Files.createDirectory(zipfs.getPath("META-INF"));
+ Path manifest = zipfs.getPath("META-INF/MANIFEST.MF");
+ Files.write(manifest, "Manifest-Version: 1.0\nCreated-By: Apache Flink\nMain-Class: HelloWorld\n".getBytes());
+
+ Path content = zipfs.getPath("HelloWorld.class");
+ Files.write(content, new byte[] {
+ /* // This byte array is the byte code of the following program:
+ * public class HelloWorld {
+ * public static void main(String[] args) {
+ * System.out.println("hello out!");
+ * System.err.println("hello err!");
+ * }
+ * }
+ */
+ -54, -2, -70, -66, 0, 0, 0, 52, 0, 39, 10, 0, 8, 0, 22, 9, 0, 23, 0, 24, 8, 0, 25, 10, 0, 26, 0, 27, 9, 0, 23, 0, 28, 8, 0, 29, 7, 0, 30, 7, 0, 31, 1, 0, 6, 60, 105, 110, 105, 116, 62, 1, 0, 3, 40, 41, 86, 1, 0, 4, 67, 111, 100, 101, 1, 0, 15, 76, 105, 110, 101, 78, 117, 109, 98, 101, 114, 84, 97, 98, 108, 101, 1, 0, 18, 76, 111, 99, 97, 108, 86, 97, 114, 105, 97, 98, 108, 101, 84, 97, 98, 108, 101, 1, 0, 4, 116, 104, 105, 115, 1, 0, 12, 76, 72, 101, 108, 108, 111, 87, 111, 114, 108, [...]
+ });
+ }
+ }
+}
diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/handlers/JarRunHandlerTest.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/handlers/JarRunHandlerTest.java
deleted file mode 100644
index fb80c92..0000000
--- a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/handlers/JarRunHandlerTest.java
+++ /dev/null
@@ -1,114 +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.webmonitor.handlers;
-
-import org.apache.flink.api.common.time.Time;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.RestOptions;
-import org.apache.flink.configuration.WebOptions;
-import org.apache.flink.runtime.rest.RestClient;
-import org.apache.flink.runtime.rest.RestClientConfiguration;
-import org.apache.flink.runtime.rest.util.RestClientException;
-import org.apache.flink.runtime.testingUtils.TestingUtils;
-import org.apache.flink.runtime.testutils.MiniClusterResource;
-import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
-import org.apache.flink.testutils.junit.category.AlsoRunWithLegacyScheduler;
-import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.TestLogger;
-
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-import org.junit.rules.TemporaryFolder;
-
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.Optional;
-
-import static org.hamcrest.CoreMatchers.containsString;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.junit.Assert.assertTrue;
-
-/**
- * Tests for the {@link JarRunHandler}.
- */
-@Category(AlsoRunWithLegacyScheduler.class)
-public class JarRunHandlerTest extends TestLogger {
-
- @ClassRule
- public static final TemporaryFolder TMP = new TemporaryFolder();
-
- @Test
- public void testRunJar() throws Exception {
- Path uploadDir = TMP.newFolder().toPath();
-
- Path actualUploadDir = uploadDir.resolve("flink-web-upload");
- Files.createDirectory(actualUploadDir);
-
- Path emptyJar = actualUploadDir.resolve("empty.jar");
- Files.createFile(emptyJar);
-
- Configuration config = new Configuration();
- config.setString(WebOptions.UPLOAD_DIR, uploadDir.toString());
-
- MiniClusterResource clusterResource = new MiniClusterResource(
- new MiniClusterResourceConfiguration.Builder()
- .setConfiguration(config)
- .setNumberTaskManagers(1)
- .setNumberSlotsPerTaskManager(1)
- .build());
- clusterResource.before();
-
- try {
- Configuration clientConfig = clusterResource.getClientConfiguration();
- RestClient client = new RestClient(RestClientConfiguration.fromConfiguration(clientConfig), TestingUtils.defaultExecutor());
-
- try {
- JarRunHeaders headers = JarRunHeaders.getInstance();
- JarRunMessageParameters parameters = headers.getUnresolvedMessageParameters();
- parameters.jarIdPathParameter.resolve(emptyJar.getFileName().toString());
-
- String host = clientConfig.getString(RestOptions.ADDRESS);
- int port = clientConfig.getInteger(RestOptions.PORT);
-
- try {
- client.sendRequest(host, port, headers, parameters, new JarRunRequestBody())
- .get();
- } catch (Exception e) {
- Optional<RestClientException> expected = ExceptionUtils.findThrowable(e, RestClientException.class);
- if (expected.isPresent()) {
- // implies the job was actually submitted
- assertTrue(expected.get().getMessage().contains("ProgramInvocationException"));
- // original cause is preserved in stack trace
- assertThat(expected.get().getMessage(), containsString("ZipException: zip file is empty"));
- // implies the jar was registered for the job graph (otherwise the jar name would not occur in the exception)
- // implies the jar was uploaded (otherwise the file would not be found at all)
- assertTrue(expected.get().getMessage().contains("empty.jar"));
- } else {
- throw e;
- }
- }
- } finally {
- client.shutdown(Time.milliseconds(10));
- }
- } finally {
- clusterResource.after();
- }
- }
-}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/optimizer/jsonplan/DumpCompiledPlanTest.java b/flink-tests/src/test/java/org/apache/flink/test/optimizer/jsonplan/DumpCompiledPlanTest.java
index 0f31481..04eb024 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/optimizer/jsonplan/DumpCompiledPlanTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/optimizer/jsonplan/DumpCompiledPlanTest.java
@@ -104,7 +104,7 @@ public class DumpCompiledPlanTest extends CompilerTestBase {
.setArguments(args)
.build();
- final Pipeline pipeline = PackagedProgramUtils.getPipelineFromProgram(program, 1);
+ final Pipeline pipeline = PackagedProgramUtils.getPipelineFromProgram(program, 1, true);
assertTrue(pipeline instanceof Plan);
diff --git a/flink-tests/src/test/java/org/apache/flink/test/optimizer/jsonplan/PreviewPlanDumpTest.java b/flink-tests/src/test/java/org/apache/flink/test/optimizer/jsonplan/PreviewPlanDumpTest.java
index 0591e41..24d4ab0 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/optimizer/jsonplan/PreviewPlanDumpTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/optimizer/jsonplan/PreviewPlanDumpTest.java
@@ -107,7 +107,7 @@ public class PreviewPlanDumpTest extends CompilerTestBase {
.setArguments(args)
.build();
- final Pipeline pipeline = PackagedProgramUtils.getPipelineFromProgram(program, 1);
+ final Pipeline pipeline = PackagedProgramUtils.getPipelineFromProgram(program, 1, true);
assertTrue(pipeline instanceof Plan);
diff --git a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
index 7a3cfd8..81fef14 100644
--- a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
+++ b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnConfigurationITCase.java
@@ -105,7 +105,7 @@ public class YarnConfigurationITCase extends YarnTestBase {
final File streamingWordCountFile = getTestJarPath("WindowJoin.jar");
final PackagedProgram packagedProgram = PackagedProgram.newBuilder().setJarFile(streamingWordCountFile).build();
- final JobGraph jobGraph = PackagedProgramUtils.createJobGraph(packagedProgram, configuration, 1);
+ final JobGraph jobGraph = PackagedProgramUtils.createJobGraph(packagedProgram, configuration, 1, false);
try {
final ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder()