You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by uc...@apache.org on 2016/07/18 15:05:40 UTC
[2/2] flink git commit: [FLINK-3713] [clients,
runtime] Use user code class loader when disposing savepoint
[FLINK-3713] [clients, runtime] Use user code class loader when disposing savepoint
Disposing savepoints via the JobManager fails for state handles or descriptors,
which contain user classes (for example custom folding state or RocksDB handles).
With this change, the user can provide the job JAR when disposing a savepoint in
order to use the user code class loader of that job. The JAR is optional, hence
not breaking the CLI API.
This closes #2083.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b67e508d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b67e508d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b67e508d
Branch: refs/heads/master
Commit: b67e508d02906cdfa48205a30dc9aa2b5d7222ae
Parents: bd8d7f5
Author: Ufuk Celebi <uc...@apache.org>
Authored: Wed Jun 8 10:59:24 2016 +0200
Committer: Ufuk Celebi <uc...@apache.org>
Committed: Mon Jul 18 17:04:09 2016 +0200
----------------------------------------------------------------------
docs/apis/cli.md | 49 +--
.../org/apache/flink/client/CliFrontend.java | 93 +++--
.../flink/client/cli/CliFrontendParser.java | 7 +-
.../flink/client/cli/SavepointOptions.java | 15 +-
.../flink/client/program/PackagedProgram.java | 6 +-
.../flink/client/CliFrontendSavepointTest.java | 141 ++++++--
.../webmonitor/handlers/JarRunHandler.java | 2 +-
.../apache/flink/runtime/blob/BlobClient.java | 111 +++++-
.../apache/flink/runtime/blob/BlobServer.java | 1 +
.../checkpoint/CheckpointCoordinator.java | 6 +-
.../CheckpointCoordinatorDeActivator.java | 2 +-
.../runtime/checkpoint/CompletedCheckpoint.java | 12 +-
.../flink/runtime/checkpoint/KeyGroupState.java | 8 +-
.../runtime/checkpoint/PendingCheckpoint.java | 21 +-
.../SavepointCoordinatorDeActivator.java | 2 +-
.../runtime/checkpoint/SavepointStore.java | 12 +-
.../StandaloneCompletedCheckpointStore.java | 4 +-
.../flink/runtime/checkpoint/SubtaskState.java | 8 +-
.../flink/runtime/checkpoint/TaskState.java | 2 +-
.../apache/flink/runtime/client/JobClient.java | 45 +--
.../flink/runtime/client/JobClientActor.java | 2 +-
.../apache/flink/runtime/jobgraph/JobGraph.java | 23 +-
.../flink/runtime/jobmanager/JobManager.scala | 21 +-
.../runtime/messages/JobManagerMessages.scala | 9 +-
.../flink/runtime/blob/BlobClientTest.java | 24 ++
.../CompletedCheckpointStoreTest.java | 2 +-
flink-tests/pom.xml | 19 ++
.../assembly/test-custom_kv_state-assembly.xml | 38 +++
.../test/checkpointing/SavepointITCase.java | 5 +-
.../test/classloading/ClassLoaderITCase.java | 335 ++++++++++++++-----
.../classloading/jar/CustomKvStateProgram.java | 120 +++++++
31 files changed, 858 insertions(+), 287 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/docs/apis/cli.md
----------------------------------------------------------------------
diff --git a/docs/apis/cli.md b/docs/apis/cli.md
index 511862c..c272413 100644
--- a/docs/apis/cli.md
+++ b/docs/apis/cli.md
@@ -31,9 +31,9 @@ and connects by default to the running Flink master (JobManager) that was
started from the same installation directory.
A prerequisite to using the command line interface is that the Flink
-master (JobManager) has been started (via
-`<flink-home>/bin/start-local.sh` or
-`<flink-home>/bin/start-cluster.sh`) or that a YARN environment is
+master (JobManager) has been started (via
+`<flink-home>/bin/start-local.sh` or
+`<flink-home>/bin/start-cluster.sh`) or that a YARN environment is
available.
The command line can be used to
@@ -116,11 +116,11 @@ The command line can be used to
- Stop a job (streaming jobs only):
./bin/flink stop <jobID>
-
-
+
+
The difference between cancelling and stopping a (streaming) job is the following:
-On a cancel call, the operators in a job immediately receive a `cancel()` method call to cancel them as
+On a cancel call, the operators in a job immediately receive a `cancel()` method call to cancel them as
soon as possible.
If operators are not not stopping after the cancel call, Flink will start interrupting the thread periodically
until it stops.
@@ -142,7 +142,7 @@ This allows the job to finish processing all inflight data.
Returns the path of the created savepoint. You need this path to restore and dispose savepoints.
-#### **Restore a savepoint**:
+#### **Restore a savepoint**
{% highlight bash %}
./bin/flink run -s <savepointPath> ...
@@ -150,7 +150,7 @@ Returns the path of the created savepoint. You need this path to restore and dis
The run command has a savepoint flag to submit a job, which restores its state from a savepoint. The savepoint path is returned by the savepoint trigger command.
-#### **Dispose a savepoint**:
+#### **Dispose a savepoint**
{% highlight bash %}
./bin/flink savepoint -d <savepointPath>
@@ -158,6 +158,14 @@ The run command has a savepoint flag to submit a job, which restores its state f
Disposes the savepoint at the given path. The savepoint path is returned by the savepoint trigger command.
+If you use custom state instances (for example custom reducing state or RocksDB state), you have to specify the path to the program JAR with which the savepoint was triggered in order to dispose the savepoint with the user code class loader:
+
+{% highlight bash %}
+./bin/flink savepoint -d <savepointPath> -j <jarFile>
+{% endhighlight %}
+
+Otherwise, you will run into a `ClassNotFoundException`.
+
## Usage
The command line syntax is as follows:
@@ -301,19 +309,14 @@ guarantees for a stop request.
Action "savepoint" triggers savepoints for a running job or disposes existing ones.
- Syntax: savepoint [OPTIONS] <Job ID>
- "savepoint" action options:
- -d,--dispose <savepointPath> Disposes an existing savepoint.
- -m,--jobmanager <host:port> Address of the JobManager (master) to which
- to connect. Specify 'yarn-cluster' as the
- JobManager to deploy a YARN cluster for the
- job. Use this flag to connect to a different
- JobManager than the one specified in the
- configuration.
- Additional arguments if -m yarn-cluster is set:
- -yid <yarnApplicationId> YARN application ID of Flink YARN session to
- connect to. Must not be set if JobManager HA
- is used. In this case, JobManager RPC
- location is automatically retrieved from
- Zookeeper.
+ Syntax: savepoint [OPTIONS] <Job ID>
+ "savepoint" action options:
+ -d,--dispose <arg> Path of savepoint to dispose.
+ -j,--jarfile <jarfile> Flink program JAR file.
+ -m,--jobmanager <host:port> Address of the JobManager (master) to which
+ to connect. Use this flag to connect to a
+ different JobManager than the one specified
+ in the configuration.
+ Options for yarn-cluster mode:
+ -yid,--yarnapplicationId <arg> Attach to running YARN session
~~~
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java b/flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java
index 404cc21..8d36715 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java
@@ -44,6 +44,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
import org.apache.flink.optimizer.DataStatistics;
import org.apache.flink.optimizer.Optimizer;
import org.apache.flink.optimizer.costs.DefaultCostEstimator;
@@ -52,6 +53,8 @@ import org.apache.flink.optimizer.plan.OptimizedPlan;
import org.apache.flink.optimizer.plan.StreamingPlan;
import org.apache.flink.optimizer.plandump.PlanJSONDumpGenerator;
import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.blob.BlobClient;
+import org.apache.flink.runtime.blob.BlobKey;
import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.runtime.instance.ActorGateway;
import org.apache.flink.runtime.jobgraph.JobStatus;
@@ -67,10 +70,9 @@ import org.apache.flink.runtime.security.SecurityUtils;
import org.apache.flink.runtime.util.EnvironmentInformation;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.StringUtils;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-
+import scala.Option;
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;
@@ -604,11 +606,9 @@ public class CliFrontend {
SavepointOptions options;
try {
options = CliFrontendParser.parseSavepointCommand(args);
- }
- catch (CliArgsException e) {
+ } catch (CliArgsException e) {
return handleArgException(e);
- }
- catch (Throwable t) {
+ } catch (Throwable t) {
return handleError(t);
}
@@ -620,9 +620,8 @@ public class CliFrontend {
if (options.isDispose()) {
// Discard
- return disposeSavepoint(options, options.getDisposeSavepointPath());
- }
- else {
+ return disposeSavepoint(options);
+ } else {
// Trigger
String[] cleanedArgs = options.getArgs();
JobID jobId;
@@ -631,14 +630,12 @@ public class CliFrontend {
String jobIdString = cleanedArgs[0];
try {
jobId = new JobID(StringUtils.hexStringToByte(jobIdString));
- }
- catch (Exception e) {
- return handleError(new IllegalArgumentException(
+ } catch (Exception e) {
+ return handleArgException(new IllegalArgumentException(
"Error: The value for the Job ID is not a valid ID."));
}
- }
- else {
- return handleError(new IllegalArgumentException(
+ } else {
+ return handleArgException(new IllegalArgumentException(
"Error: The value for the Job ID is not a valid ID. " +
"Specify a Job ID to trigger a savepoint."));
}
@@ -693,35 +690,77 @@ public class CliFrontend {
* Sends a {@link org.apache.flink.runtime.messages.JobManagerMessages.DisposeSavepoint}
* message to the job manager.
*/
- private int disposeSavepoint(SavepointOptions options, String savepointPath) {
+ private int disposeSavepoint(SavepointOptions options) {
try {
+ String savepointPath = options.getSavepointPath();
+ if (savepointPath == null) {
+ throw new IllegalArgumentException("Missing required argument: savepoint path. " +
+ "Usage: bin/flink savepoint -d <savepoint-path>");
+ }
+
+ String jarFile = options.getJarFilePath();
+
ActorGateway jobManager = getJobManagerGateway(options);
- logAndSysout("Disposing savepoint '" + savepointPath + "'.");
- Future<Object> response = jobManager.ask(new DisposeSavepoint(savepointPath), clientTimeout);
+
+ List<BlobKey> blobKeys = null;
+ if (jarFile != null) {
+ logAndSysout("Disposing savepoint '" + savepointPath + "' with JAR " + jarFile + ".");
+
+ List<File> libs = null;
+ try {
+ libs = PackagedProgram.extractContainedLibraries(new File(jarFile).toURI().toURL());
+ if (!libs.isEmpty()) {
+ List<Path> libPaths = new ArrayList<>(libs.size());
+ for (File f : libs) {
+ libPaths.add(new Path(f.toURI()));
+ }
+
+ logAndSysout("Uploading JAR files.");
+ LOG.debug("JAR files: " + libPaths);
+ blobKeys = BlobClient.uploadJarFiles(jobManager, clientTimeout, libPaths);
+ LOG.debug("Blob keys: " + blobKeys.toString());
+ }
+ } finally {
+ if (libs != null) {
+ PackagedProgram.deleteExtractedLibraries(libs);
+ }
+ }
+ } else {
+ logAndSysout("Disposing savepoint '" + savepointPath + "'.");
+ }
+
+ Object msg = new DisposeSavepoint(savepointPath, Option.apply(blobKeys));
+ Future<Object> response = jobManager.ask(msg, clientTimeout);
Object result;
try {
logAndSysout("Waiting for response...");
result = Await.result(response, clientTimeout);
- }
- catch (Exception e) {
+ } catch (Exception e) {
throw new Exception("Disposing the savepoint with path" + savepointPath + " failed.", e);
}
if (result.getClass() == JobManagerMessages.getDisposeSavepointSuccess().getClass()) {
logAndSysout("Savepoint '" + savepointPath + "' disposed.");
return 0;
- }
- else if (result instanceof DisposeSavepointFailure) {
+ } else if (result instanceof DisposeSavepointFailure) {
DisposeSavepointFailure failure = (DisposeSavepointFailure) result;
- throw failure.cause();
- }
- else {
+
+ if (failure.cause() instanceof ClassNotFoundException) {
+ throw new ClassNotFoundException("Savepoint disposal failed, because of a " +
+ "missing class. This is most likely caused by a custom state " +
+ "instance, which cannot be disposed without the user code class " +
+ "loader. Please provide the program jar with which you have created " +
+ "the savepoint via -j <JAR> for disposal.",
+ failure.cause().getCause());
+ } else {
+ throw failure.cause();
+ }
+ } else {
throw new IllegalStateException("Unknown JobManager response of type " +
result.getClass());
}
- }
- catch (Throwable t) {
+ } catch (Throwable t) {
return handleError(t);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
index d479d35..819c241 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontendParser.java
@@ -73,7 +73,7 @@ public class CliFrontendParser {
"Path to a savepoint to reset the job back to (for example file:///flink/savepoint-1537).");
static final Option SAVEPOINT_DISPOSE_OPTION = new Option("d", "dispose", true,
- "Disposes an existing savepoint.");
+ "Path of savepoint to dispose.");
// list specific options
static final Option RUNNING_OPTION = new Option("r", "running", false,
@@ -112,9 +112,6 @@ public class CliFrontendParser {
SAVEPOINT_PATH_OPTION.setRequired(false);
SAVEPOINT_PATH_OPTION.setArgName("savepointPath");
-
- SAVEPOINT_DISPOSE_OPTION.setRequired(false);
- SAVEPOINT_DISPOSE_OPTION.setArgName("savepointPath");
}
private static final Options RUN_OPTIONS = getRunOptions(buildGeneralOptions(new Options()));
@@ -195,6 +192,7 @@ public class CliFrontendParser {
private static Options getSavepointOptions(Options options) {
options = getJobManagerAddressOption(options);
options.addOption(SAVEPOINT_DISPOSE_OPTION);
+ options.addOption(JAR_OPTION);
return addCustomCliOptions(options, false);
}
@@ -234,6 +232,7 @@ public class CliFrontendParser {
private static Options getSavepointOptionsWithoutDeprecatedOptions(Options options) {
options = getJobManagerAddressOption(options);
options.addOption(SAVEPOINT_DISPOSE_OPTION);
+ options.addOption(JAR_OPTION);
return options;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-clients/src/main/java/org/apache/flink/client/cli/SavepointOptions.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/cli/SavepointOptions.java b/flink-clients/src/main/java/org/apache/flink/client/cli/SavepointOptions.java
index 10af76e..305b0b4 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/cli/SavepointOptions.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/cli/SavepointOptions.java
@@ -19,6 +19,7 @@ package org.apache.flink.client.cli;
import org.apache.commons.cli.CommandLine;
+import static org.apache.flink.client.cli.CliFrontendParser.JAR_OPTION;
import static org.apache.flink.client.cli.CliFrontendParser.SAVEPOINT_DISPOSE_OPTION;
/**
@@ -29,12 +30,14 @@ public class SavepointOptions extends CommandLineOptions {
private final String[] args;
private boolean dispose;
private String disposeSavepointPath;
+ private String jarFile;
public SavepointOptions(CommandLine line) {
super(line);
- this.args = line.getArgs();
- this.dispose = line.hasOption(SAVEPOINT_DISPOSE_OPTION.getOpt());
- this.disposeSavepointPath = line.getOptionValue(SAVEPOINT_DISPOSE_OPTION.getOpt());
+ args = line.getArgs();
+ dispose = line.hasOption(SAVEPOINT_DISPOSE_OPTION.getOpt());
+ disposeSavepointPath = line.getOptionValue(SAVEPOINT_DISPOSE_OPTION.getOpt());
+ jarFile = line.getOptionValue(JAR_OPTION.getOpt());
}
public String[] getArgs() {
@@ -45,7 +48,11 @@ public class SavepointOptions extends CommandLineOptions {
return dispose;
}
- public String getDisposeSavepointPath() {
+ public String getSavepointPath() {
return disposeSavepointPath;
}
+
+ public String getJarFilePath() {
+ return jarFile;
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgram.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgram.java b/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgram.java
index 40092d8..bff8e3e 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgram.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/PackagedProgram.java
@@ -189,7 +189,7 @@ public class PackagedProgram {
}
// now that we have an entry point, we can extract the nested jar files (if any)
- this.extractedTempLibraries = extractContainedLibaries(jarFileUrl);
+ this.extractedTempLibraries = extractContainedLibraries(jarFileUrl);
this.classpaths = classpaths;
this.userCodeClassLoader = JobWithJars.buildUserCodeClassLoader(getAllLibraries(), classpaths, getClass().getClassLoader());
@@ -642,7 +642,7 @@ public class PackagedProgram {
* @return The file names of the extracted temporary files.
* @throws ProgramInvocationException Thrown, if the extraction process failed.
*/
- private static List<File> extractContainedLibaries(URL jarFile) throws ProgramInvocationException {
+ public static List<File> extractContainedLibraries(URL jarFile) throws ProgramInvocationException {
Random rnd = new Random();
@@ -741,7 +741,7 @@ public class PackagedProgram {
}
}
- private static void deleteExtractedLibraries(List<File> tempLibraries) {
+ public static void deleteExtractedLibraries(List<File> tempLibraries) {
for (File f : tempLibraries) {
f.delete();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-clients/src/test/java/org/apache/flink/client/CliFrontendSavepointTest.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/test/java/org/apache/flink/client/CliFrontendSavepointTest.java b/flink-clients/src/test/java/org/apache/flink/client/CliFrontendSavepointTest.java
index 13c895c..d9e7075 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/CliFrontendSavepointTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/CliFrontendSavepointTest.java
@@ -18,20 +18,38 @@
package org.apache.flink.client;
+import akka.dispatch.Futures;
import org.apache.flink.api.common.JobID;
import org.apache.flink.client.cli.CommandLineOptions;
+import org.apache.flink.runtime.blob.BlobKey;
import org.apache.flink.runtime.instance.ActorGateway;
import org.apache.flink.runtime.messages.JobManagerMessages;
+import org.junit.Rule;
import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
import org.mockito.Mockito;
+import scala.Option;
+import scala.concurrent.Future;
import scala.concurrent.Promise;
import scala.concurrent.duration.FiniteDuration;
import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.FileOutputStream;
import java.io.PrintStream;
-
+import java.util.List;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipOutputStream;
+
+import static org.apache.flink.runtime.messages.JobManagerMessages.DisposeSavepoint;
+import static org.apache.flink.runtime.messages.JobManagerMessages.DisposeSavepointFailure;
+import static org.apache.flink.runtime.messages.JobManagerMessages.TriggerSavepoint;
+import static org.apache.flink.runtime.messages.JobManagerMessages.TriggerSavepointFailure;
+import static org.apache.flink.runtime.messages.JobManagerMessages.TriggerSavepointSuccess;
+import static org.apache.flink.runtime.messages.JobManagerMessages.getDisposeSavepointSuccess;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -43,6 +61,9 @@ public class CliFrontendSavepointTest {
private static PrintStream stdErr;
private static ByteArrayOutputStream buffer;
+ @Rule
+ public TemporaryFolder tmp = new TemporaryFolder();
+
// ------------------------------------------------------------------------
// Trigger savepoint
// ------------------------------------------------------------------------
@@ -58,14 +79,13 @@ public class CliFrontendSavepointTest {
Promise<Object> triggerResponse = new scala.concurrent.impl.Promise.DefaultPromise<>();
when(jobManager.ask(
- Mockito.eq(new JobManagerMessages.TriggerSavepoint(jobId)),
- Mockito.any(FiniteDuration.class)))
+ Mockito.eq(new TriggerSavepoint(jobId)),
+ any(FiniteDuration.class)))
.thenReturn(triggerResponse.future());
String savepointPath = "expectedSavepointPath";
- triggerResponse.success(new JobManagerMessages
- .TriggerSavepointSuccess(jobId, savepointPath));
+ triggerResponse.success(new TriggerSavepointSuccess(jobId, savepointPath));
CliFrontend frontend = new MockCliFrontend(
CliFrontendTestUtils.getConfigDir(), jobManager);
@@ -75,8 +95,8 @@ public class CliFrontendSavepointTest {
assertEquals(0, returnCode);
verify(jobManager, times(1)).ask(
- Mockito.eq(new JobManagerMessages.TriggerSavepoint(jobId)),
- Mockito.any(FiniteDuration.class));
+ Mockito.eq(new TriggerSavepoint(jobId)),
+ any(FiniteDuration.class));
assertTrue(buffer.toString().contains("expectedSavepointPath"));
}
@@ -96,14 +116,13 @@ public class CliFrontendSavepointTest {
Promise<Object> triggerResponse = new scala.concurrent.impl.Promise.DefaultPromise<>();
when(jobManager.ask(
- Mockito.eq(new JobManagerMessages.TriggerSavepoint(jobId)),
- Mockito.any(FiniteDuration.class)))
+ Mockito.eq(new TriggerSavepoint(jobId)),
+ any(FiniteDuration.class)))
.thenReturn(triggerResponse.future());
Exception testException = new Exception("expectedTestException");
- triggerResponse.success(new JobManagerMessages
- .TriggerSavepointFailure(jobId, testException));
+ triggerResponse.success(new TriggerSavepointFailure(jobId, testException));
CliFrontend frontend = new MockCliFrontend(
CliFrontendTestUtils.getConfigDir(), jobManager);
@@ -113,8 +132,8 @@ public class CliFrontendSavepointTest {
assertTrue(returnCode != 0);
verify(jobManager, times(1)).ask(
- Mockito.eq(new JobManagerMessages.TriggerSavepoint(jobId)),
- Mockito.any(FiniteDuration.class));
+ Mockito.eq(new TriggerSavepoint(jobId)),
+ any(FiniteDuration.class));
assertTrue(buffer.toString().contains("expectedTestException"));
}
@@ -152,8 +171,8 @@ public class CliFrontendSavepointTest {
Promise<Object> triggerResponse = new scala.concurrent.impl.Promise.DefaultPromise<>();
when(jobManager.ask(
- Mockito.eq(new JobManagerMessages.TriggerSavepoint(jobId)),
- Mockito.any(FiniteDuration.class)))
+ Mockito.eq(new TriggerSavepoint(jobId)),
+ any(FiniteDuration.class)))
.thenReturn(triggerResponse.future());
triggerResponse.success("UNKNOWN RESPONSE");
@@ -166,8 +185,8 @@ public class CliFrontendSavepointTest {
assertTrue(returnCode != 0);
verify(jobManager, times(1)).ask(
- Mockito.eq(new JobManagerMessages.TriggerSavepoint(jobId)),
- Mockito.any(FiniteDuration.class));
+ Mockito.eq(new TriggerSavepoint(jobId)),
+ any(FiniteDuration.class));
String errMsg = buffer.toString();
assertTrue(errMsg.contains("IllegalStateException"));
@@ -193,10 +212,10 @@ public class CliFrontendSavepointTest {
Promise<Object> triggerResponse = new scala.concurrent.impl.Promise.DefaultPromise<>();
when(jobManager.ask(
- Mockito.eq(new JobManagerMessages.DisposeSavepoint(savepointPath)),
- Mockito.any(FiniteDuration.class))).thenReturn(triggerResponse.future());
+ Mockito.eq(new DisposeSavepoint(savepointPath, Option.<List<BlobKey>>empty())),
+ any(FiniteDuration.class))).thenReturn(triggerResponse.future());
- triggerResponse.success(JobManagerMessages.getDisposeSavepointSuccess());
+ triggerResponse.success(getDisposeSavepointSuccess());
CliFrontend frontend = new MockCliFrontend(
CliFrontendTestUtils.getConfigDir(), jobManager);
@@ -206,8 +225,8 @@ public class CliFrontendSavepointTest {
assertEquals(0, returnCode);
verify(jobManager, times(1)).ask(
- Mockito.eq(new JobManagerMessages.DisposeSavepoint(savepointPath)),
- Mockito.any(FiniteDuration.class));
+ Mockito.eq(new DisposeSavepoint(savepointPath, Option.<List<BlobKey>>empty())),
+ any(FiniteDuration.class));
String outMsg = buffer.toString();
assertTrue(outMsg.contains(savepointPath));
@@ -218,6 +237,65 @@ public class CliFrontendSavepointTest {
}
}
+ /**
+ * Tests that a disposal failure due a ClassNotFoundException triggers a
+ * note about the JAR option.
+ */
+ @Test
+ public void testDisposeClassNotFoundException() throws Exception {
+ replaceStdOutAndStdErr();
+
+ try {
+ Future<Object> classNotFoundFailure = Futures
+ .<Object>successful(new DisposeSavepointFailure(new ClassNotFoundException("Test exception")));
+
+ ActorGateway jobManager = mock(ActorGateway.class);
+ when(jobManager.ask(any(DisposeSavepoint.class), any(FiniteDuration.class)))
+ .thenReturn(classNotFoundFailure);
+
+ CliFrontend frontend = new MockCliFrontend(CliFrontendTestUtils.getConfigDir(), jobManager);
+
+ String[] parameters = { "-d", "any-path" };
+
+ int returnCode = frontend.savepoint(parameters);
+ assertTrue(returnCode != 0);
+
+ String out = buffer.toString();
+ assertTrue(out.contains("Please provide the program jar with which you have created " +
+ "the savepoint via -j <JAR> for disposal"));
+ } finally {
+ restoreStdOutAndStdErr();
+ }
+ }
+
+ /**
+ * Tests disposal with a JAR file.
+ */
+ @Test
+ public void testDisposeWithJar() throws Exception {
+ replaceStdOutAndStdErr();
+
+ try {
+ ActorGateway jobManager = mock(ActorGateway.class);
+ when(jobManager.ask(any(DisposeSavepoint.class), any(FiniteDuration.class)))
+ .thenReturn(Futures.successful(JobManagerMessages.getDisposeSavepointSuccess()));
+
+ CliFrontend frontend = new MockCliFrontend(CliFrontendTestUtils.getConfigDir(), jobManager);
+
+ // Fake JAR file
+ File f = tmp.newFile();
+ ZipOutputStream out = new ZipOutputStream(new FileOutputStream(f));
+ out.close();
+
+ String[] parameters = { "-d", "any-path", "-j", f.getAbsolutePath() };
+
+ int returnCode = frontend.savepoint(parameters);
+ assertEquals(0, returnCode);
+ } finally {
+ restoreStdOutAndStdErr();
+ }
+ }
+
@Test
public void testDisposeSavepointFailure() throws Exception {
replaceStdOutAndStdErr();
@@ -229,14 +307,13 @@ public class CliFrontendSavepointTest {
Promise<Object> triggerResponse = new scala.concurrent.impl.Promise.DefaultPromise<>();
when(jobManager.ask(
- Mockito.eq(new JobManagerMessages.DisposeSavepoint(savepointPath)),
- Mockito.any(FiniteDuration.class)))
+ Mockito.eq(new DisposeSavepoint(savepointPath, Option.<List<BlobKey>>empty())),
+ any(FiniteDuration.class)))
.thenReturn(triggerResponse.future());
Exception testException = new Exception("expectedTestException");
- triggerResponse.success(new JobManagerMessages
- .DisposeSavepointFailure(testException));
+ triggerResponse.success(new DisposeSavepointFailure(testException));
CliFrontend frontend = new MockCliFrontend(
CliFrontendTestUtils.getConfigDir(), jobManager);
@@ -246,8 +323,8 @@ public class CliFrontendSavepointTest {
assertTrue(returnCode != 0);
verify(jobManager, times(1)).ask(
- Mockito.eq(new JobManagerMessages.DisposeSavepoint(savepointPath)),
- Mockito.any(FiniteDuration.class));
+ Mockito.eq(new DisposeSavepoint(savepointPath, Option.<List<BlobKey>>empty())),
+ any(FiniteDuration.class));
assertTrue(buffer.toString().contains("expectedTestException"));
}
@@ -267,8 +344,8 @@ public class CliFrontendSavepointTest {
Promise<Object> triggerResponse = new scala.concurrent.impl.Promise.DefaultPromise<>();
when(jobManager.ask(
- Mockito.eq(new JobManagerMessages.DisposeSavepoint(savepointPath)),
- Mockito.any(FiniteDuration.class)))
+ Mockito.eq(new DisposeSavepoint(savepointPath, Option.<List<BlobKey>>empty())),
+ any(FiniteDuration.class)))
.thenReturn(triggerResponse.future());
triggerResponse.success("UNKNOWN RESPONSE");
@@ -281,8 +358,8 @@ public class CliFrontendSavepointTest {
assertTrue(returnCode != 0);
verify(jobManager, times(1)).ask(
- Mockito.eq(new JobManagerMessages.DisposeSavepoint(savepointPath)),
- Mockito.any(FiniteDuration.class));
+ Mockito.eq(new DisposeSavepoint(savepointPath, Option.<List<BlobKey>>empty())),
+ any(FiniteDuration.class));
String errMsg = buffer.toString();
assertTrue(errMsg.contains("IllegalStateException"));
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/handlers/JarRunHandler.java
----------------------------------------------------------------------
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 d7b8e72..b9e773a 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
@@ -49,7 +49,7 @@ public class JarRunHandler extends JarActionHandler {
try {
Tuple2<JobGraph, ClassLoader> graph = getJobGraphAndClassLoader(pathParams, queryParams);
try {
- JobClient.uploadJarFiles(graph.f0, jobManager, timeout);
+ graph.f0.uploadUserJars(jobManager, timeout);
} catch (IOException e) {
throw new ProgramInvocationException("Failed to upload jar files to the job manager", e);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobClient.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobClient.java b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobClient.java
index 1c76e08..fa4c08c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobClient.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobClient.java
@@ -18,6 +18,20 @@
package org.apache.flink.runtime.blob;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.instance.ActorGateway;
+import org.apache.flink.runtime.messages.JobManagerMessages;
+import org.apache.flink.util.InstantiationUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Option;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
+
import java.io.Closeable;
import java.io.EOFException;
import java.io.FileNotFoundException;
@@ -27,26 +41,23 @@ import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.security.MessageDigest;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
-import org.apache.flink.util.InstantiationUtil;
-import org.slf4j.LoggerFactory;
-import org.slf4j.Logger;
-
-import org.apache.flink.api.common.JobID;
-
+import static org.apache.flink.runtime.blob.BlobServerProtocol.BUFFER_SIZE;
import static org.apache.flink.runtime.blob.BlobServerProtocol.CONTENT_ADDRESSABLE;
+import static org.apache.flink.runtime.blob.BlobServerProtocol.DELETE_OPERATION;
+import static org.apache.flink.runtime.blob.BlobServerProtocol.GET_OPERATION;
import static org.apache.flink.runtime.blob.BlobServerProtocol.JOB_ID_SCOPE;
+import static org.apache.flink.runtime.blob.BlobServerProtocol.MAX_KEY_LENGTH;
import static org.apache.flink.runtime.blob.BlobServerProtocol.NAME_ADDRESSABLE;
+import static org.apache.flink.runtime.blob.BlobServerProtocol.PUT_OPERATION;
+import static org.apache.flink.runtime.blob.BlobServerProtocol.RETURN_ERROR;
+import static org.apache.flink.runtime.blob.BlobServerProtocol.RETURN_OKAY;
import static org.apache.flink.runtime.blob.BlobUtils.readFully;
import static org.apache.flink.runtime.blob.BlobUtils.readLength;
import static org.apache.flink.runtime.blob.BlobUtils.writeLength;
-import static org.apache.flink.runtime.blob.BlobServerProtocol.BUFFER_SIZE;
-import static org.apache.flink.runtime.blob.BlobServerProtocol.DELETE_OPERATION;
-import static org.apache.flink.runtime.blob.BlobServerProtocol.GET_OPERATION;
-import static org.apache.flink.runtime.blob.BlobServerProtocol.PUT_OPERATION;
-import static org.apache.flink.runtime.blob.BlobServerProtocol.MAX_KEY_LENGTH;
-import static org.apache.flink.runtime.blob.BlobServerProtocol.RETURN_OKAY;
-import static org.apache.flink.runtime.blob.BlobServerProtocol.RETURN_ERROR;
/**
* The BLOB client can communicate with the BLOB server and either upload (PUT), download (GET),
@@ -656,6 +667,80 @@ public final class BlobClient implements Closeable {
}
}
+ /**
+ * Retrieves the {@link BlobServer} address from the JobManager and uploads
+ * the JAR files to it.
+ *
+ * @param jobManager Server address of the {@link BlobServer}
+ * @param askTimeout Ask timeout for blob server address retrieval
+ * @param jars List of JAR files to upload
+ * @throws IOException Thrown if the address retrieval or upload fails
+ */
+ public static List<BlobKey> uploadJarFiles(
+ ActorGateway jobManager,
+ FiniteDuration askTimeout,
+ List<Path> jars) throws IOException {
+
+ if (jars.isEmpty()) {
+ return Collections.emptyList();
+ } else {
+ Object msg = JobManagerMessages.getRequestBlobManagerPort();
+ Future<Object> futureBlobPort = jobManager.ask(msg, askTimeout);
+
+ try {
+ // Retrieve address
+ Object result = Await.result(futureBlobPort, askTimeout);
+ if (result instanceof Integer) {
+ int port = (Integer) result;
+
+ Option<String> jmHost = jobManager.actor().path().address().host();
+ String jmHostname = jmHost.isDefined() ? jmHost.get() : "localhost";
+ InetSocketAddress serverAddress = new InetSocketAddress(jmHostname, port);
+
+ // Now, upload
+ return uploadJarFiles(serverAddress, jars);
+ } else {
+ throw new Exception("Expected port number (int) as answer, received " + result);
+ }
+ } catch (Exception e) {
+ throw new IOException("Could not retrieve the JobManager's blob port.", e);
+ }
+ }
+ }
+
+ /**
+ * Uploads the JAR files to a {@link BlobServer} at the given address.
+ *
+ * @param serverAddress Server address of the {@link BlobServer}
+ * @param jars List of JAR files to upload
+ * @throws IOException Thrown if the upload fails
+ */
+ public static List<BlobKey> uploadJarFiles(InetSocketAddress serverAddress, List<Path> jars) throws IOException {
+ if (jars.isEmpty()) {
+ return Collections.emptyList();
+ } else {
+ List<BlobKey> blobKeys = new ArrayList<>();
+
+ try (BlobClient blobClient = new BlobClient(serverAddress)) {
+ for (final Path jar : jars) {
+ final FileSystem fs = jar.getFileSystem();
+ FSDataInputStream is = null;
+ try {
+ is = fs.open(jar);
+ final BlobKey key = blobClient.put(is);
+ blobKeys.add(key);
+ } finally {
+ if (is != null) {
+ is.close();
+ }
+ }
+ }
+ }
+
+ return blobKeys;
+ }
+ }
+
// --------------------------------------------------------------------------------------------
// Miscellaneous
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java
index 2120b2f..ed3a68b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java
@@ -431,4 +431,5 @@ public class BlobServer extends Thread implements BlobService {
return new ArrayList<BlobServerConnection>(activeConnections);
}
}
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
index c6b2a77..c21ebc0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
@@ -732,7 +732,7 @@ public class CheckpointCoordinator {
recentPendingCheckpoints.addLast(id);
}
- private void dropSubsumedCheckpoints(long timestamp) {
+ private void dropSubsumedCheckpoints(long timestamp) throws Exception {
Iterator<Map.Entry<Long, PendingCheckpoint>> entries = pendingCheckpoints.entrySet().iterator();
while (entries.hasNext()) {
PendingCheckpoint p = entries.next().getValue();
@@ -924,7 +924,7 @@ public class CheckpointCoordinator {
// Periodic scheduling of checkpoints
// --------------------------------------------------------------------------------------------
- public void startCheckpointScheduler() {
+ public void startCheckpointScheduler() throws Exception {
synchronized (lock) {
if (shutdown) {
throw new IllegalArgumentException("Checkpoint coordinator is shut down");
@@ -947,7 +947,7 @@ public class CheckpointCoordinator {
}
}
- public void stopCheckpointScheduler() {
+ public void stopCheckpointScheduler() throws Exception {
synchronized (lock) {
triggerRequestQueued = false;
periodicScheduling = false;
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorDeActivator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorDeActivator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorDeActivator.java
index 115d1db..7e26f71 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorDeActivator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorDeActivator.java
@@ -45,7 +45,7 @@ public class CheckpointCoordinatorDeActivator extends FlinkUntypedActor {
}
@Override
- public void handleMessage(Object message) {
+ public void handleMessage(Object message) throws Exception {
if (message instanceof ExecutionGraphMessages.JobStatusChanged) {
JobStatus status = ((ExecutionGraphMessages.JobStatusChanged) message).newJobStatus();
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java
index f899ae1..3e1db02 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CompletedCheckpoint.java
@@ -97,12 +97,14 @@ public class CompletedCheckpoint implements Serializable {
// --------------------------------------------------------------------------------------------
- public void discard(ClassLoader userClassLoader) {
- for (TaskState state: taskStates.values()) {
- state.discard(userClassLoader);
+ public void discard(ClassLoader userClassLoader) throws Exception {
+ try {
+ for (TaskState state : taskStates.values()) {
+ state.discard(userClassLoader);
+ }
+ } finally {
+ taskStates.clear();
}
-
- taskStates.clear();
}
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/KeyGroupState.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/KeyGroupState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/KeyGroupState.java
index f510151..eb358b6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/KeyGroupState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/KeyGroupState.java
@@ -62,12 +62,8 @@ public class KeyGroupState implements Serializable {
return stateSize;
}
- public void discard(ClassLoader classLoader) {
- try {
- keyGroupState.deserializeValue(classLoader).discardState();
- } catch (Exception e) {
- LOG.warn("Failed to discard checkpoint state: " + this, e);
- }
+ public void discard(ClassLoader classLoader) throws Exception {
+ keyGroupState.deserializeValue(classLoader).discardState();
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
index 850440e..8c9069c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/PendingCheckpoint.java
@@ -53,6 +53,7 @@ public class PendingCheckpoint {
private int numAcknowledgedTasks;
private boolean discarded;
+ private ClassLoader userClassLoader;
// --------------------------------------------------------------------------------------------
@@ -109,7 +110,7 @@ public class PendingCheckpoint {
return discarded;
}
- public CompletedCheckpoint toCompletedCheckpoint() {
+ public CompletedCheckpoint toCompletedCheckpoint() throws Exception {
synchronized (lock) {
if (discarded) {
throw new IllegalStateException("pending checkpoint is discarded");
@@ -194,21 +195,25 @@ public class PendingCheckpoint {
/**
* Discards the pending checkpoint, releasing all held resources.
*/
- public void discard(ClassLoader userClassLoader) {
+ public void discard(ClassLoader userClassLoader) throws Exception {
+ this.userClassLoader = userClassLoader;
dispose(userClassLoader, true);
}
- private void dispose(ClassLoader userClassLoader, boolean releaseState) {
+ private void dispose(ClassLoader userClassLoader, boolean releaseState) throws Exception {
synchronized (lock) {
discarded = true;
numAcknowledgedTasks = -1;
- if (releaseState) {
- for (TaskState taskState : taskStates.values()) {
- taskState.discard(userClassLoader);
+ try {
+ if (releaseState) {
+ for (TaskState taskState : taskStates.values()) {
+ taskState.discard(userClassLoader);
+ }
}
+ } finally {
+ taskStates.clear();
+ notYetAcknowledgedTasks.clear();
}
- taskStates.clear();
- notYetAcknowledgedTasks.clear();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinatorDeActivator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinatorDeActivator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinatorDeActivator.java
index ca2b3ff..56a3738 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinatorDeActivator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinatorDeActivator.java
@@ -45,7 +45,7 @@ public class SavepointCoordinatorDeActivator extends FlinkUntypedActor {
}
@Override
- public void handleMessage(Object message) {
+ public void handleMessage(Object message) throws Exception {
if (message instanceof ExecutionGraphMessages.JobStatusChanged) {
JobStatus status = ((ExecutionGraphMessages.JobStatusChanged) message).newJobStatus();
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointStore.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointStore.java
index 4bdbee0..2ab4d00 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointStore.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointStore.java
@@ -32,15 +32,17 @@ public class SavepointStore implements StateStore<CompletedCheckpoint> {
public void start() {
}
- public void stop() {
+ public void stop() throws Exception {
if (stateStore instanceof HeapStateStore) {
HeapStateStore<CompletedCheckpoint> heapStateStore = (HeapStateStore<CompletedCheckpoint>) stateStore;
- for (CompletedCheckpoint savepoint : heapStateStore.getAll()) {
- savepoint.discard(ClassLoader.getSystemClassLoader());
+ try {
+ for (CompletedCheckpoint savepoint : heapStateStore.getAll()) {
+ savepoint.discard(ClassLoader.getSystemClassLoader());
+ }
+ } finally {
+ heapStateStore.clearAll();
}
-
- heapStateStore.clearAll();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StandaloneCompletedCheckpointStore.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StandaloneCompletedCheckpointStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StandaloneCompletedCheckpointStore.java
index c56f89e..8e33860 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StandaloneCompletedCheckpointStore.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StandaloneCompletedCheckpointStore.java
@@ -67,7 +67,7 @@ class StandaloneCompletedCheckpointStore implements CompletedCheckpointStore {
}
@Override
- public void addCheckpoint(CompletedCheckpoint checkpoint) {
+ public void addCheckpoint(CompletedCheckpoint checkpoint) throws Exception {
checkpoints.addLast(checkpoint);
if (checkpoints.size() > maxNumberOfCheckpointsToRetain) {
checkpoints.removeFirst().discard(userClassLoader);
@@ -90,7 +90,7 @@ class StandaloneCompletedCheckpointStore implements CompletedCheckpointStore {
}
@Override
- public void discardAllCheckpoints() {
+ public void discardAllCheckpoints() throws Exception {
for (CompletedCheckpoint checkpoint : checkpoints) {
checkpoint.discard(userClassLoader);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SubtaskState.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SubtaskState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SubtaskState.java
index 04ba8a5..9fcafe8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SubtaskState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SubtaskState.java
@@ -81,12 +81,8 @@ public class SubtaskState implements Serializable {
return duration;
}
- public void discard(ClassLoader userClassLoader) {
- try {
- state.deserializeValue(userClassLoader).discardState();
- } catch (Exception e) {
- LOG.warn("Failed to discard checkpoint state: " + this, e);
- }
+ public void discard(ClassLoader userClassLoader) throws Exception {
+ state.deserializeValue(userClassLoader).discardState();
}
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskState.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskState.java
index 2d57021..ac4503d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/TaskState.java
@@ -142,7 +142,7 @@ public class TaskState implements Serializable {
return kvStates.size();
}
- public void discard(ClassLoader classLoader) {
+ public void discard(ClassLoader classLoader) throws Exception {
for (SubtaskState subtaskState : subtaskStates.values()) {
subtaskState.discard(classLoader);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java b/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java
index 46a432e..c0e0d08 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClient.java
@@ -25,7 +25,6 @@ import akka.actor.PoisonPill;
import akka.actor.Props;
import akka.pattern.Patterns;
import akka.util.Timeout;
-
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.Configuration;
@@ -37,11 +36,9 @@ import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
import org.apache.flink.runtime.messages.JobClientMessages;
import org.apache.flink.runtime.messages.JobManagerMessages;
import org.apache.flink.runtime.util.SerializedThrowable;
-
import org.apache.flink.util.NetUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-
import scala.Option;
import scala.Some;
import scala.Tuple2;
@@ -51,7 +48,6 @@ import scala.concurrent.duration.FiniteDuration;
import java.io.IOException;
import java.net.InetAddress;
-import java.net.InetSocketAddress;
import java.util.concurrent.TimeoutException;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -208,7 +204,7 @@ public class JobClient {
LOG.info("Checking and uploading JAR files");
try {
- JobClient.uploadJarFiles(jobGraph, jobManagerGateway, timeout);
+ jobGraph.uploadUserJars(jobManagerGateway, timeout);
}
catch (IOException e) {
throw new JobSubmissionException(jobGraph.getJobID(),
@@ -262,43 +258,4 @@ public class JobClient {
throw new JobExecutionException(jobGraph.getJobID(), "Unexpected response from JobManager: " + result);
}
}
-
- /**
- * Uploads the specified jar files of the [[JobGraph]] jobGraph to the BlobServer of the
- * JobManager. The respective port is retrieved from the JobManager. This function issues a
- * blocking call.
- *
- * @param jobGraph Flink job containing the information about the required jars
- * @param jobManagerGateway Gateway to the JobManager.
- * @param timeout Timeout for futures
- * @throws IOException Thrown, if the file upload to the JobManager failed.
- */
- public static void uploadJarFiles(JobGraph jobGraph, ActorGateway jobManagerGateway, FiniteDuration timeout)
- throws IOException {
-
- if (jobGraph.hasUsercodeJarFiles()) {
- Future<Object> futureBlobPort = jobManagerGateway.ask(
- JobManagerMessages.getRequestBlobManagerPort(),
- timeout);
-
- int port;
- try {
- Object result = Await.result(futureBlobPort, timeout);
- if (result instanceof Integer) {
- port = (Integer) result;
- } else {
- throw new Exception("Expected port number (int) as answer, received " + result);
- }
- }
- catch (Exception e) {
- throw new IOException("Could not retrieve the JobManager's blob port.", e);
- }
-
- Option<String> jmHost = jobManagerGateway.actor().path().address().host();
- String jmHostname = jmHost.isDefined() ? jmHost.get() : "localhost";
- InetSocketAddress serverAddress = new InetSocketAddress(jmHostname, port);
-
- jobGraph.uploadRequiredJarFiles(serverAddress);
- }
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClientActor.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClientActor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClientActor.java
index e51946f..2b3138a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClientActor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/client/JobClientActor.java
@@ -352,7 +352,7 @@ public class JobClientActor extends FlinkUntypedActor implements LeaderRetrieval
LOG.info("Upload jar files to job manager {}.", jobManager.path());
try {
- JobClient.uploadJarFiles(jobGraph, jobManagerGateway, timeout);
+ jobGraph.uploadUserJars(jobManagerGateway, timeout);
} catch (IOException exception) {
getSelf().tell(
decorateMessage(new JobManagerMessages.JobResultFailure(
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
index b3e3739..f825d5b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
@@ -27,21 +27,23 @@ import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.blob.BlobClient;
import org.apache.flink.runtime.blob.BlobKey;
+import org.apache.flink.runtime.instance.ActorGateway;
import org.apache.flink.runtime.jobgraph.tasks.JobSnapshottingSettings;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.SerializedValue;
+import scala.concurrent.duration.FiniteDuration;
import java.io.IOException;
import java.io.Serializable;
import java.net.InetSocketAddress;
import java.net.URL;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
-import java.util.Collections;
import java.util.Set;
/**
@@ -541,6 +543,25 @@ public class JobGraph implements Serializable {
return maxParallelism;
}
+ /**
+ * Uploads the previously added user JAR files to the job manager through
+ * the job manager's BLOB server. The respective port is retrieved from the
+ * JobManager. This function issues a blocking call.
+ *
+ * @param jobManager JobManager actor gateway
+ * @param askTimeout Ask timeout
+ * @throws IOException Thrown, if the file upload to the JobManager failed.
+ */
+ public void uploadUserJars(ActorGateway jobManager, FiniteDuration askTimeout) throws IOException {
+ List<BlobKey> blobKeys = BlobClient.uploadJarFiles(jobManager, askTimeout, userJars);
+
+ for (BlobKey blobKey : blobKeys) {
+ if (!userJarBlobKeys.contains(blobKey)) {
+ userJarBlobKeys.add(blobKey);
+ }
+ }
+ }
+
@Override
public String toString() {
return "JobGraph(jobId: " + jobID + ")";
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
index fc066b1..3f3c02e 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
@@ -745,7 +745,7 @@ class JobManager(
sender() ! TriggerSavepointFailure(jobId, new IllegalArgumentException("Unknown job."))
}
- case DisposeSavepoint(savepointPath) =>
+ case DisposeSavepoint(savepointPath, blobKeys) =>
val senderRef = sender()
future {
try {
@@ -755,8 +755,23 @@ class JobManager(
log.debug(s"$savepoint")
- // Discard the associated checkpoint
- savepoint.discard(getClass.getClassLoader)
+ if (blobKeys.isDefined) {
+ // We don't need a real ID here for the library cache manager
+ val jid = new JobID()
+
+ try {
+ libraryCacheManager.registerJob(jid, blobKeys.get, java.util.Collections.emptyList())
+ val classLoader = libraryCacheManager.getClassLoader(jid)
+
+ // Discard with user code loader
+ savepoint.discard(classLoader)
+ } finally {
+ libraryCacheManager.unregisterJob(jid)
+ }
+ } else {
+ // Discard with system class loader
+ savepoint.discard(getClass.getClassLoader)
+ }
// Dispose the savepoint
savepointStore.disposeState(savepointPath)
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobManagerMessages.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobManagerMessages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobManagerMessages.scala
index c949b4c..14f72b0 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobManagerMessages.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/JobManagerMessages.scala
@@ -23,6 +23,7 @@ import java.util.UUID
import akka.actor.ActorRef
import org.apache.flink.api.common.JobID
import org.apache.flink.runtime.akka.ListeningBehaviour
+import org.apache.flink.runtime.blob.BlobKey
import org.apache.flink.runtime.client.{JobStatusMessage, SerializedJobExecutionResult}
import org.apache.flink.runtime.executiongraph.{ExecutionAttemptID, ExecutionGraph}
import org.apache.flink.runtime.instance.{Instance, InstanceID}
@@ -447,8 +448,14 @@ object JobManagerMessages {
* Disposes a savepoint.
*
* @param savepointPath The path of the savepoint to dispose.
+ * @param blobKeys BLOB keys if a user program JAR was uploaded for disposal.
+ * This is required when we dispose state which contains
+ * custom state instances (e.g. reducing state, rocksDB state).
*/
- case class DisposeSavepoint(savepointPath: String) extends RequiresLeaderSessionID
+ case class DisposeSavepoint(
+ savepointPath: String,
+ blobKeys: Option[java.util.List[BlobKey]] = None)
+ extends RequiresLeaderSessionID
/** Response after a successful savepoint dispose. */
case object DisposeSavepointSuccess
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobClientTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobClientTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobClientTest.java
index db74bbd..ccdd3a1 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobClientTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobClientTest.java
@@ -29,9 +29,12 @@ import java.io.IOException;
import java.io.InputStream;
import java.net.InetSocketAddress;
import java.security.MessageDigest;
+import java.util.Collections;
+import java.util.List;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.core.fs.Path;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -379,4 +382,25 @@ public class BlobClientTest {
fail(e.getMessage());
}
}
+
+ /**
+ * Tests the static {@link BlobClient#uploadJarFiles(InetSocketAddress, List)} helper.
+ */
+ @Test
+ public void testUploadJarFilesHelper() throws Exception {
+ final File testFile = File.createTempFile("testfile", ".dat");
+ testFile.deleteOnExit();
+ prepareTestFile(testFile);
+
+ InetSocketAddress serverAddress = new InetSocketAddress("localhost", BLOB_SERVER.getPort());
+
+ List<BlobKey> blobKeys = BlobClient.uploadJarFiles(serverAddress, Collections.singletonList(new Path(testFile.toURI())));
+
+ assertEquals(1, blobKeys.size());
+
+ try (BlobClient blobClient = new BlobClient(serverAddress)) {
+ InputStream is = blobClient.get(blobKeys.get(0));
+ validateGet(is, testFile);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStoreTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStoreTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStoreTest.java
index 0276a1f..f273b63 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStoreTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStoreTest.java
@@ -242,7 +242,7 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
}
@Override
- public void discard(ClassLoader userClassLoader) {
+ public void discard(ClassLoader userClassLoader) throws Exception {
super.discard(userClassLoader);
if (!isDiscarded) {
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-tests/pom.xml
----------------------------------------------------------------------
diff --git a/flink-tests/pom.xml b/flink-tests/pom.xml
index 7ced2d0..52c3fd2 100644
--- a/flink-tests/pom.xml
+++ b/flink-tests/pom.xml
@@ -459,6 +459,25 @@ under the License.
</descriptors>
</configuration>
</execution>
+ <execution>
+ <id>create-custom_kv_state-jar</id>
+ <phase>process-test-classes</phase>
+ <goals>
+ <goal>single</goal>
+ </goals>
+ <configuration>
+ <archive>
+ <manifest>
+ <mainClass>org.apache.flink.test.classloading.jar.CustomKvStateProgram</mainClass>
+ </manifest>
+ </archive>
+ <finalName>custom_kv_state</finalName>
+ <attach>false</attach>
+ <descriptors>
+ <descriptor>src/test/assembly/test-custom_kv_state-assembly.xml</descriptor>
+ </descriptors>
+ </configuration>
+ </execution>
</executions>
</plugin>
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-tests/src/test/assembly/test-custom_kv_state-assembly.xml
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/assembly/test-custom_kv_state-assembly.xml b/flink-tests/src/test/assembly/test-custom_kv_state-assembly.xml
new file mode 100644
index 0000000..6e60ddc
--- /dev/null
+++ b/flink-tests/src/test/assembly/test-custom_kv_state-assembly.xml
@@ -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.
+
+-->
+
+<assembly>
+ <id>test-jar</id>
+ <formats>
+ <format>jar</format>
+ </formats>
+ <includeBaseDirectory>false</includeBaseDirectory>
+ <fileSets>
+ <fileSet>
+ <directory>${project.build.testOutputDirectory}</directory>
+ <outputDirectory>/</outputDirectory>
+ <!--modify/add include to match your package(s) -->
+ <includes>
+ <include>org/apache/flink/test/classloading/jar/CustomKvStateProgram.class</include>
+ <include>org/apache/flink/test/classloading/jar/CustomKvStateProgram*.class</include>
+ </includes>
+ </fileSet>
+ </fileSets>
+</assembly>
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
index 89761ff..c4e0f42 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
@@ -30,6 +30,7 @@ import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.blob.BlobKey;
import org.apache.flink.runtime.checkpoint.CompletedCheckpoint;
import org.apache.flink.runtime.checkpoint.SavepointStoreFactory;
import org.apache.flink.runtime.checkpoint.SubtaskState;
@@ -70,6 +71,7 @@ import org.junit.Rule;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import scala.Option;
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.Deadline;
@@ -345,7 +347,8 @@ public class SavepointITCase extends TestLogger {
LOG.info("Disposing savepoint " + savepointPath + ".");
Future<Object> disposeFuture = jobManager.ask(
- new DisposeSavepoint(savepointPath), deadline.timeLeft());
+ new DisposeSavepoint(savepointPath, Option.<List<BlobKey>>empty()),
+ deadline.timeLeft());
errMsg = "Failed to dispose savepoint " + savepointPath + ".";
Object resp = Await.result(disposeFuture, deadline.timeLeft());
http://git-wip-us.apache.org/repos/asf/flink/blob/b67e508d/flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java b/flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java
index 04abdf8..9e24944 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/classloading/ClassLoaderITCase.java
@@ -18,23 +18,52 @@
package org.apache.flink.test.classloading;
-import java.io.File;
-
+import org.apache.flink.api.common.JobID;
import org.apache.flink.client.program.PackagedProgram;
+import org.apache.flink.client.program.ProgramInvocationException;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.blob.BlobClient;
+import org.apache.flink.runtime.blob.BlobKey;
+import org.apache.flink.runtime.checkpoint.SavepointStoreFactory;
+import org.apache.flink.runtime.client.JobStatusMessage;
+import org.apache.flink.runtime.instance.ActorGateway;
+import org.apache.flink.runtime.messages.JobManagerMessages;
+import org.apache.flink.runtime.messages.JobManagerMessages.DisposeSavepoint;
+import org.apache.flink.runtime.messages.JobManagerMessages.DisposeSavepointFailure;
+import org.apache.flink.runtime.messages.JobManagerMessages.RunningJobsStatus;
+import org.apache.flink.runtime.messages.JobManagerMessages.TriggerSavepoint;
+import org.apache.flink.runtime.messages.JobManagerMessages.TriggerSavepointSuccess;
import org.apache.flink.runtime.state.filesystem.FsStateBackendFactory;
+import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages.WaitForAllVerticesToBeRunning;
import org.apache.flink.test.testdata.KMeansData;
import org.apache.flink.test.util.ForkableFlinkMiniCluster;
-import org.junit.Rule;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Option;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.Deadline;
+import scala.concurrent.duration.FiniteDuration;
+
+import java.io.File;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.fail;
public class ClassLoaderITCase {
+ private static final Logger LOG = LoggerFactory.getLogger(ClassLoaderITCase.class);
+
private static final String INPUT_SPLITS_PROG_JAR_FILE = "customsplit-test-jar.jar";
private static final String STREAMING_INPUT_SPLITS_PROG_JAR_FILE = "streaming-customsplit-test-jar.jar";
@@ -47,114 +76,240 @@ public class ClassLoaderITCase {
private static final String USERCODETYPE_JAR_PATH = "usercodetype-test-jar.jar";
- @Rule
- public TemporaryFolder folder = new TemporaryFolder();
+ private static final String CUSTOM_KV_STATE_JAR_PATH = "custom_kv_state-test-jar.jar";
+
+ public static final TemporaryFolder FOLDER = new TemporaryFolder();
+
+ private static ForkableFlinkMiniCluster testCluster;
+
+ private static int parallelism;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ FOLDER.create();
+
+ Configuration config = new Configuration();
+ config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 2);
+ config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 2);
+ parallelism = 4;
+
+ // we need to use the "filesystem" state backend to ensure FLINK-2543 is not happening again.
+ config.setString(ConfigConstants.STATE_BACKEND, "filesystem");
+ config.setString(FsStateBackendFactory.CHECKPOINT_DIRECTORY_URI_CONF_KEY,
+ FOLDER.newFolder().getAbsoluteFile().toURI().toString());
+
+ // Savepoint path
+ config.setString(SavepointStoreFactory.SAVEPOINT_BACKEND_KEY, "filesystem");
+ config.setString(SavepointStoreFactory.SAVEPOINT_DIRECTORY_KEY,
+ FOLDER.newFolder().getAbsoluteFile().toURI().toString());
+
+ testCluster = new ForkableFlinkMiniCluster(config, false);
+ testCluster.start();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ if (testCluster != null) {
+ testCluster.shutdown();
+ }
+
+ FOLDER.delete();
+ }
@Test
public void testJobsWithCustomClassLoader() {
try {
- Configuration config = new Configuration();
- config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 2);
- config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 2);
+ int port = testCluster.getLeaderRPCPort();
+
+ PackagedProgram inputSplitTestProg = new PackagedProgram(
+ new File(INPUT_SPLITS_PROG_JAR_FILE),
+ new String[] { INPUT_SPLITS_PROG_JAR_FILE,
+ "", // classpath
+ "localhost",
+ String.valueOf(port),
+ "4" // parallelism
+ });
+ inputSplitTestProg.invokeInteractiveModeForExecution();
- // we need to use the "filesystem" state backend to ensure FLINK-2543 is not happening again.
- config.setString(ConfigConstants.STATE_BACKEND, "filesystem");
- config.setString(FsStateBackendFactory.CHECKPOINT_DIRECTORY_URI_CONF_KEY,
- folder.newFolder().getAbsoluteFile().toURI().toString());
+ PackagedProgram streamingInputSplitTestProg = new PackagedProgram(
+ new File(STREAMING_INPUT_SPLITS_PROG_JAR_FILE),
+ new String[] { STREAMING_INPUT_SPLITS_PROG_JAR_FILE,
+ "localhost",
+ String.valueOf(port),
+ "4" // parallelism
+ });
+ streamingInputSplitTestProg.invokeInteractiveModeForExecution();
- ForkableFlinkMiniCluster testCluster = new ForkableFlinkMiniCluster(config, false);
+ String classpath = new File(INPUT_SPLITS_PROG_JAR_FILE).toURI().toURL().toString();
+ PackagedProgram inputSplitTestProg2 = new PackagedProgram(new File(INPUT_SPLITS_PROG_JAR_FILE),
+ new String[] { "",
+ classpath, // classpath
+ "localhost",
+ String.valueOf(port),
+ "4" // parallelism
+ });
+ inputSplitTestProg2.invokeInteractiveModeForExecution();
- testCluster.start();
+ // regular streaming job
+ PackagedProgram streamingProg = new PackagedProgram(
+ new File(STREAMING_PROG_JAR_FILE),
+ new String[] {
+ STREAMING_PROG_JAR_FILE,
+ "localhost",
+ String.valueOf(port)
+ });
+ streamingProg.invokeInteractiveModeForExecution();
+ // checkpointed streaming job with custom classes for the checkpoint (FLINK-2543)
+ // the test also ensures that user specific exceptions are serializable between JobManager <--> JobClient.
try {
- int port = testCluster.getLeaderRPCPort();
-
- PackagedProgram inputSplitTestProg = new PackagedProgram(
- new File(INPUT_SPLITS_PROG_JAR_FILE),
- new String[] { INPUT_SPLITS_PROG_JAR_FILE,
- "", // classpath
- "localhost",
- String.valueOf(port),
- "4" // parallelism
- });
- inputSplitTestProg.invokeInteractiveModeForExecution();
-
- PackagedProgram streamingInputSplitTestProg = new PackagedProgram(
- new File(STREAMING_INPUT_SPLITS_PROG_JAR_FILE),
- new String[] { STREAMING_INPUT_SPLITS_PROG_JAR_FILE,
- "localhost",
- String.valueOf(port),
- "4" // parallelism
- });
- streamingInputSplitTestProg.invokeInteractiveModeForExecution();
-
- String classpath = new File(INPUT_SPLITS_PROG_JAR_FILE).toURI().toURL().toString();
- PackagedProgram inputSplitTestProg2 = new PackagedProgram(new File(INPUT_SPLITS_PROG_JAR_FILE),
- new String[] { "",
- classpath, // classpath
- "localhost",
- String.valueOf(port),
- "4" // parallelism
- } );
- inputSplitTestProg2.invokeInteractiveModeForExecution();
-
- // regular streaming job
- PackagedProgram streamingProg = new PackagedProgram(
- new File(STREAMING_PROG_JAR_FILE),
+ PackagedProgram streamingCheckpointedProg = new PackagedProgram(
+ new File(STREAMING_CHECKPOINTED_PROG_JAR_FILE),
new String[] {
- STREAMING_PROG_JAR_FILE,
+ STREAMING_CHECKPOINTED_PROG_JAR_FILE,
"localhost",
- String.valueOf(port)
- });
- streamingProg.invokeInteractiveModeForExecution();
+ String.valueOf(port) });
+ streamingCheckpointedProg.invokeInteractiveModeForExecution();
+ } catch (Exception e) {
+ // we can not access the SuccessException here when executing the tests with maven, because its not available in the jar.
+ assertEquals("Program should terminate with a 'SuccessException'",
+ "org.apache.flink.test.classloading.jar.CheckpointedStreamingProgram.SuccessException",
+ e.getCause().getCause().getClass().getCanonicalName());
+ }
- // checkpointed streaming job with custom classes for the checkpoint (FLINK-2543)
- // the test also ensures that user specific exceptions are serializable between JobManager <--> JobClient.
- try {
- PackagedProgram streamingCheckpointedProg = new PackagedProgram(
- new File(STREAMING_CHECKPOINTED_PROG_JAR_FILE),
- new String[] {
- STREAMING_CHECKPOINTED_PROG_JAR_FILE,
- "localhost",
- String.valueOf(port)});
- streamingCheckpointedProg.invokeInteractiveModeForExecution();
- }
- catch (Exception e) {
- // we can not access the SuccessException here when executing the tests with maven, because its not available in the jar.
- assertEquals("Program should terminate with a 'SuccessException'",
- "org.apache.flink.test.classloading.jar.CheckpointedStreamingProgram.SuccessException",
- e.getCause().getCause().getClass().getCanonicalName());
- }
+ PackagedProgram kMeansProg = new PackagedProgram(
+ new File(KMEANS_JAR_PATH),
+ new String[] { KMEANS_JAR_PATH,
+ "localhost",
+ String.valueOf(port),
+ "4", // parallelism
+ KMeansData.DATAPOINTS,
+ KMeansData.INITIAL_CENTERS,
+ "25"
+ });
+ kMeansProg.invokeInteractiveModeForExecution();
- PackagedProgram kMeansProg = new PackagedProgram(
- new File(KMEANS_JAR_PATH),
- new String[] { KMEANS_JAR_PATH,
- "localhost",
- String.valueOf(port),
- "4", // parallelism
- KMeansData.DATAPOINTS,
- KMeansData.INITIAL_CENTERS,
- "25"
- });
- kMeansProg.invokeInteractiveModeForExecution();
-
- // test FLINK-3633
- PackagedProgram userCodeTypeProg = new PackagedProgram(
+ // test FLINK-3633
+ final PackagedProgram userCodeTypeProg = new PackagedProgram(
new File(USERCODETYPE_JAR_PATH),
new String[] { USERCODETYPE_JAR_PATH,
+ "localhost",
+ String.valueOf(port),
+ });
+
+ userCodeTypeProg.invokeInteractiveModeForExecution();
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ /**
+ * Tests disposal of a savepoint, which contains custom user code KvState.
+ */
+ @Test
+ public void testDisposeSavepointWithCustomKvState() throws Exception {
+ Deadline deadline = new FiniteDuration(100, TimeUnit.SECONDS).fromNow();
+
+ int port = testCluster.getLeaderRPCPort();
+
+ File checkpointDir = FOLDER.newFolder();
+ File outputDir = FOLDER.newFolder();
+
+ final PackagedProgram program = new PackagedProgram(
+ new File(CUSTOM_KV_STATE_JAR_PATH),
+ new String[] {
+ CUSTOM_KV_STATE_JAR_PATH,
"localhost",
String.valueOf(port),
- });
+ String.valueOf(parallelism),
+ checkpointDir.toURI().toString(),
+ "5000",
+ outputDir.toURI().toString()
+ });
- userCodeTypeProg.invokeInteractiveModeForExecution();
+ // Execute detached
+ Thread invokeThread = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ program.invokeInteractiveModeForExecution();
+ } catch (ProgramInvocationException ignored) {
+ ignored.printStackTrace();
+ }
}
- finally {
- testCluster.shutdown();
+ });
+
+ LOG.info("Starting program invoke thread");
+ invokeThread.start();
+
+ // The job ID
+ JobID jobId = null;
+
+ ActorGateway jm = testCluster.getLeaderGateway(deadline.timeLeft());
+
+ LOG.info("Waiting for job status running.");
+
+ // Wait for running job
+ while (jobId == null && deadline.hasTimeLeft()) {
+ Future<Object> jobsFuture = jm.ask(JobManagerMessages.getRequestRunningJobsStatus(), deadline.timeLeft());
+ RunningJobsStatus runningJobs = (RunningJobsStatus) Await.result(jobsFuture, deadline.timeLeft());
+
+ for (JobStatusMessage runningJob : runningJobs.getStatusMessages()) {
+ jobId = runningJob.getJobId();
+ LOG.info("Job running. ID: " + jobId);
+ break;
+ }
+
+ // Retry if job is not available yet
+ if (jobId == null) {
+ Thread.sleep(100);
}
}
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
+
+ LOG.info("Wait for all tasks to be running.");
+ Future<Object> allRunning = jm.ask(new WaitForAllVerticesToBeRunning(jobId), deadline.timeLeft());
+ Await.ready(allRunning, deadline.timeLeft());
+ LOG.info("All tasks are running.");
+
+ // Trigger savepoint
+ String savepointPath = null;
+ for (int i = 0; i < 20; i++) {
+ LOG.info("Triggering savepoint (" + (i+1) + "/20.");
+ Future<Object> savepointFuture = jm.ask(new TriggerSavepoint(jobId), deadline.timeLeft());
+ Object savepointResponse = Await.result(savepointFuture, deadline.timeLeft());
+
+ if (savepointResponse.getClass() == TriggerSavepointSuccess.class) {
+ savepointPath = ((TriggerSavepointSuccess) savepointResponse).savepointPath();
+ LOG.info("Triggered savepoint. Path: " + savepointPath);
+ } else if (savepointResponse.getClass() == JobManagerMessages.TriggerSavepointFailure.class) {
+ Throwable cause = ((JobManagerMessages.TriggerSavepointFailure) savepointResponse).cause();
+ LOG.info("Failed to trigger savepoint. Retrying...", cause);
+ // This can fail if the operators are not opened yet
+ Thread.sleep(500);
+ } else {
+ throw new IllegalStateException("Unexpected response to TriggerSavepoint");
+ }
+ }
+
+ assertNotNull(savepointPath, "Failed to trigger savepoint");
+
+ // Upload JAR
+ LOG.info("Uploading JAR " + CUSTOM_KV_STATE_JAR_PATH + " for savepoint disposal.");
+ List<BlobKey> blobKeys = BlobClient.uploadJarFiles(jm, deadline.timeLeft(), Collections.singletonList(new Path(CUSTOM_KV_STATE_JAR_PATH)));
+
+ // Dispose savepoint
+ LOG.info("Disposing savepoint at " + savepointPath);
+ Future<Object> disposeFuture = jm.ask(new DisposeSavepoint(savepointPath, Option.apply(blobKeys)), deadline.timeLeft());
+ Object disposeResponse = Await.result(disposeFuture, deadline.timeLeft());
+
+ if (disposeResponse.getClass() == JobManagerMessages.getDisposeSavepointSuccess().getClass()) {
+ // Success :-)
+ LOG.info("Disposed savepoint at " + savepointPath);
+ } else if (disposeResponse instanceof DisposeSavepointFailure) {
+ throw new IllegalStateException("Failed to dispose savepoint");
+ } else {
+ throw new IllegalStateException("Unexpected response to DisposeSavepoint");
}
}
}