You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by rm...@apache.org on 2015/11/25 16:41:17 UTC
[1/2] flink git commit: [FLINK-3000] Adds shutdown hook to clean up
lingering yarn sessions
Repository: flink
Updated Branches:
refs/heads/master 90c76ad1c -> 008060f16
[FLINK-3000] Adds shutdown hook to clean up lingering yarn sessions
This closes #1354
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/aad99f25
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/aad99f25
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/aad99f25
Branch: refs/heads/master
Commit: aad99f25cd107a8eaa24a8d407680a8186ba6460
Parents: 90c76ad
Author: Sachin Goel <sa...@gmail.com>
Authored: Sun Nov 15 06:11:30 2015 +0530
Committer: Robert Metzger <rm...@apache.org>
Committed: Wed Nov 25 16:20:37 2015 +0100
----------------------------------------------------------------------
.../org/apache/flink/client/CliFrontend.java | 8 ----
.../apache/flink/yarn/FlinkYarnClientBase.java | 39 +++++++++++++++++++-
2 files changed, 37 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/aad99f25/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 6a79677..5a57bf4 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
@@ -635,14 +635,6 @@ public class CliFrontend {
// --------------------------------------------------------------------------------------------
protected int executeProgramDetached(PackagedProgram program, Client client, int parallelism) {
- // log message for detached yarn job
- if (yarnCluster != null) {
- logAndSysout("The Flink YARN client has been started in detached mode. In order to stop " +
- "Flink on YARN, use the following command or a YARN web interface to stop it:\n" +
- "yarn application -kill " + yarnCluster.getApplicationId() + "\n" +
- "Please also note that the temporary files of the YARN session in the home directoy will not be removed.");
- }
-
LOG.info("Starting execution of program");
JobSubmissionResult result;
http://git-wip-us.apache.org/repos/asf/flink/blob/aad99f25/flink-yarn/src/main/java/org/apache/flink/yarn/FlinkYarnClientBase.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/FlinkYarnClientBase.java b/flink-yarn/src/main/java/org/apache/flink/yarn/FlinkYarnClientBase.java
index 8c4a53e..74ef5c3 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/FlinkYarnClientBase.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/FlinkYarnClientBase.java
@@ -110,7 +110,7 @@ public abstract class FlinkYarnClientBase extends AbstractFlinkYarnClient {
private Configuration conf;
private YarnClient yarnClient;
private YarnClientApplication yarnApplication;
-
+ private Thread deploymentFailureHook = new DeploymentFailureHook();
/**
* Files (usually in a distributed file system) used for the YARN session of Flink.
@@ -629,13 +629,20 @@ public abstract class FlinkYarnClientBase extends AbstractFlinkYarnClient {
appContext.setQueue(yarnQueue);
}
+ // add a hook to clean up in case deployment fails
+ Runtime.getRuntime().addShutdownHook(deploymentFailureHook);
LOG.info("Submitting application master " + appId);
yarnClient.submitApplication(appContext);
LOG.info("Waiting for the cluster to be allocated");
int waittime = 0;
loop: while( true ) {
- ApplicationReport report = yarnClient.getApplicationReport(appId);
+ ApplicationReport report;
+ try {
+ report = yarnClient.getApplicationReport(appId);
+ } catch (IOException e) {
+ throw new YarnDeploymentException("Failed to deploy the cluster: " + e.getMessage());
+ }
YarnApplicationState appState = report.getYarnApplicationState();
switch(appState) {
case FAILED:
@@ -660,6 +667,19 @@ public abstract class FlinkYarnClientBase extends AbstractFlinkYarnClient {
waittime += 1000;
Thread.sleep(1000);
}
+ // print the application id for user to cancel themselves.
+ if (isDetached()) {
+ LOG.info("The Flink YARN client has been started in detached mode. In order to stop " +
+ "Flink on YARN, use the following command or a YARN web interface to stop " +
+ "it:\nyarn application -kill " + appId + "\nPlease also note that the " +
+ "temporary files of the YARN session in the home directoy will not be removed.");
+ }
+ // since deployment was successful, remove the hook
+ try {
+ Runtime.getRuntime().removeShutdownHook(deploymentFailureHook);
+ } catch (IllegalStateException e) {
+ // we're already in the shut down hook.
+ }
// the Flink cluster is deployed in YARN. Represent cluster
return new FlinkYarnCluster(yarnClient, appId, conf, flinkConfiguration, sessionFilesDir, detached);
}
@@ -869,5 +889,20 @@ public abstract class FlinkYarnClientBase extends AbstractFlinkYarnClient {
}
}
+ private class DeploymentFailureHook extends Thread {
+ @Override
+ public void run() {
+ LOG.info("Cancelling deployment from Deployment Failure Hook");
+ failSessionDuringDeployment();
+ LOG.info("Deleting files in " + sessionFilesDir);
+ try {
+ FileSystem fs = FileSystem.get(conf);
+ fs.delete(sessionFilesDir, true);
+ fs.close();
+ } catch (IOException e) {
+ LOG.error("Failed to delete Flink Jar and conf files in HDFS", e);
+ }
+ }
+ }
}
[2/2] flink git commit: [hotfix] Register signal handler for
JobManager and TaskManager
Posted by rm...@apache.org.
[hotfix] Register signal handler for JobManager and TaskManager
This closes #1400
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/008060f1
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/008060f1
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/008060f1
Branch: refs/heads/master
Commit: 008060f16d5a850be178557d5d0dbbc52864c187
Parents: aad99f2
Author: Robert Metzger <rm...@apache.org>
Authored: Tue Nov 24 16:49:35 2015 +0100
Committer: Robert Metzger <rm...@apache.org>
Committed: Wed Nov 25 16:27:59 2015 +0100
----------------------------------------------------------------------
.../src/main/java/org/apache/flink/runtime/blob/BlobCache.java | 2 ++
.../main/java/org/apache/flink/runtime/util/SignalHandler.java | 1 +
.../scala/org/apache/flink/runtime/jobmanager/JobManager.scala | 1 +
.../scala/org/apache/flink/runtime/taskmanager/TaskManager.scala | 3 ++-
4 files changed, 6 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/008060f1/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobCache.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobCache.java b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobCache.java
index fd768a1..ede4d9b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobCache.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobCache.java
@@ -203,6 +203,8 @@ public final class BlobCache implements BlobService {
@Override
public void shutdown() {
if (shutdownRequested.compareAndSet(false, true)) {
+ LOG.info("Shutting down BlobCache");
+
// Clean up the storage directory
try {
FileUtils.deleteDirectory(storageDir);
http://git-wip-us.apache.org/repos/asf/flink/blob/008060f1/flink-runtime/src/main/java/org/apache/flink/runtime/util/SignalHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/SignalHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/SignalHandler.java
index bcd3dc4..0f85e71 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/util/SignalHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/SignalHandler.java
@@ -46,6 +46,7 @@ public class SignalHandler {
@Override
public void handle(Signal signal) {
LOG.error("RECEIVED SIGNAL " + signal.getNumber() + ": SIG" + signal.getName());
+ LOG.error("This JVM will shut down because it was killed from the outside.");
prevHandler.handle(signal);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/008060f1/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 3fad676..c4d0fbb 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
@@ -1382,6 +1382,7 @@ object JobManager {
// startup checks and logging
EnvironmentInformation.logEnvironmentInfo(LOG.logger, "JobManager", args)
EnvironmentInformation.checkJavaVersion()
+ SignalHandler.register(LOG.logger)
// parsing the command line arguments
val (configuration: Configuration,
http://git-wip-us.apache.org/repos/asf/flink/blob/008060f1/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
index 41533ea..f8f5204 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
@@ -64,7 +64,7 @@ import org.apache.flink.util.NetUtils
import org.apache.flink.runtime.process.ProcessReaper
import org.apache.flink.runtime.security.SecurityUtils
import org.apache.flink.runtime.security.SecurityUtils.FlinkSecuredRunner
-import org.apache.flink.runtime.util.{LeaderRetrievalUtils, MathUtils, EnvironmentInformation}
+import org.apache.flink.runtime.util.{SignalHandler, LeaderRetrievalUtils, MathUtils, EnvironmentInformation}
import scala.concurrent._
import scala.concurrent.duration._
@@ -1204,6 +1204,7 @@ object TaskManager {
// startup checks and logging
EnvironmentInformation.logEnvironmentInfo(LOG.logger, "TaskManager", args)
EnvironmentInformation.checkJavaVersion()
+ SignalHandler.register(LOG.logger)
val maxOpenFileHandles = EnvironmentInformation.getOpenFileHandlesLimit()
if (maxOpenFileHandles != -1) {