flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From m..@apache.org
Subject flink git commit: [FLINK-2226][YARN] fail application on failed single-job cluster job
Date Tue, 16 Jun 2015 15:56:37 GMT
Repository: flink
Updated Branches:
  refs/heads/master 1bd0af738 -> 46ad40588


[FLINK-2226][YARN] fail application on failed single-job cluster job

Failing jobs executed in the YARN cluster mode leave the application
container in the "SUCCEEDED" final state. While for long-running Flink
YARN clusters where multiple jobs are run, this is fine, for single jobs
it is appropriate to mark the application as failed.

This closes #838.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/46ad4058
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/46ad4058
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/46ad4058

Branch: refs/heads/master
Commit: 46ad40588538b4e5d9f2e50fe65af5a83f9c4797
Parents: 1bd0af7
Author: Maximilian Michels <mxm@apache.org>
Authored: Mon Jun 15 16:37:35 2015 +0200
Committer: Maximilian Michels <mxm@apache.org>
Committed: Tue Jun 16 17:55:23 2015 +0200

----------------------------------------------------------------------
 .../org/apache/flink/client/CliFrontend.java    |  8 ++---
 .../flink/client/FlinkYarnSessionCli.java       |  6 ++--
 .../org/apache/flink/client/program/Client.java |  1 +
 .../flink/runtime/jobgraph/JobStatus.java       |  2 +-
 .../runtime/yarn/AbstractFlinkYarnCluster.java  |  2 +-
 .../flink/yarn/YARNSessionFIFOITCase.java       |  2 +-
 .../org/apache/flink/yarn/FlinkYarnClient.java  |  4 +--
 .../org/apache/flink/yarn/FlinkYarnCluster.java | 32 +++++++++++++-------
 .../apache/flink/yarn/ApplicationClient.scala   |  2 +-
 .../flink/yarn/ApplicationMasterActor.scala     | 10 ++++--
 10 files changed, 43 insertions(+), 26 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/46ad4058/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 3e61a3b..e1bacde 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
@@ -265,6 +265,7 @@ public class CliFrontend {
 			return handleError(t);
 		}
 
+		int exitCode = 1;
 		try {
 			int userParallelism = options.getParallelism();
 			LOG.debug("User parallelism is set to {}", userParallelism);
@@ -276,15 +277,14 @@ public class CliFrontend {
 						"To use another parallelism, set it at the ./bin/flink client.");
 				userParallelism = client.getMaxSlots();
 			}
-			int exitCode = 0;
 
 			// check if detached per job yarn cluster is used to start flink
 			if(yarnCluster != null && yarnCluster.isDetached()) {
 				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" +
+						"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.");
-				executeProgram(program, client, userParallelism, false);
+				exitCode = executeProgram(program, client, userParallelism, false);
 			} else {
 				// regular (blocking) execution.
 				exitCode = executeProgram(program, client, userParallelism, true);
@@ -314,7 +314,7 @@ public class CliFrontend {
 		finally {
 			if (yarnCluster != null && !yarnCluster.isDetached()) {
 				logAndSysout("Shutting down YARN cluster");
-				yarnCluster.shutdown();
+				yarnCluster.shutdown(exitCode != 0);
 			}
 			if (program != null) {
 				program.deleteExtractedLibraries();

http://git-wip-us.apache.org/repos/asf/flink/blob/46ad4058/flink-clients/src/main/java/org/apache/flink/client/FlinkYarnSessionCli.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/FlinkYarnSessionCli.java
b/flink-clients/src/main/java/org/apache/flink/client/FlinkYarnSessionCli.java
index 0fa7173..c11edc7 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/FlinkYarnSessionCli.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/FlinkYarnSessionCli.java
@@ -302,7 +302,7 @@ public class FlinkYarnSessionCli {
 
 				if (yarnCluster.hasFailed()) {
 					System.err.println("The YARN cluster has failed");
-					yarnCluster.shutdown();
+					yarnCluster.shutdown(true);
 				}
 
 				// wait until CLIENT_POLLING_INTERVALL is over or the user entered something.
@@ -439,7 +439,7 @@ public class FlinkYarnSessionCli {
 
 				if (!yarnCluster.hasBeenStopped()) {
 					LOG.info("Command Line Interface requested session shutdown");
-					yarnCluster.shutdown();
+					yarnCluster.shutdown(false);
 				}
 
 				try {
@@ -458,7 +458,7 @@ public class FlinkYarnSessionCli {
 	public void stop() {
 		if (yarnCluster != null) {
 			LOG.info("Command line interface is shutting down the yarnCluster");
-			yarnCluster.shutdown();
+			yarnCluster.shutdown(false);
 		}
 	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/46ad4058/flink-clients/src/main/java/org/apache/flink/client/program/Client.java
----------------------------------------------------------------------
diff --git a/flink-clients/src/main/java/org/apache/flink/client/program/Client.java b/flink-clients/src/main/java/org/apache/flink/client/program/Client.java
index e219a38..c544e8d 100644
--- a/flink-clients/src/main/java/org/apache/flink/client/program/Client.java
+++ b/flink-clients/src/main/java/org/apache/flink/client/program/Client.java
@@ -318,6 +318,7 @@ public class Client {
 				ContextEnvironment.enableLocalExecution(true);
 			}
 
+			// Job id has been set in the Client passed to the ContextEnvironment
 			return new JobSubmissionResult(lastJobId);
 		}
 		else {

http://git-wip-us.apache.org/repos/asf/flink/blob/46ad4058/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobStatus.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobStatus.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobStatus.java
index 667a68e..eb7d017 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobStatus.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobStatus.java
@@ -51,7 +51,7 @@ public enum JobStatus {
 	
 	private final boolean terminalState;
 	
-	private JobStatus(boolean terminalState) {
+	JobStatus(boolean terminalState) {
 		this.terminalState = terminalState;
 	}
 	

http://git-wip-us.apache.org/repos/asf/flink/blob/46ad4058/flink-runtime/src/main/java/org/apache/flink/runtime/yarn/AbstractFlinkYarnCluster.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/yarn/AbstractFlinkYarnCluster.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/yarn/AbstractFlinkYarnCluster.java
index 398709e..c2e897f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/yarn/AbstractFlinkYarnCluster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/yarn/AbstractFlinkYarnCluster.java
@@ -30,7 +30,7 @@ public abstract class AbstractFlinkYarnCluster {
 
 	public abstract String getWebInterfaceURL();
 
-	public abstract void shutdown();
+	public abstract void shutdown(boolean failApplication);
 
 	public abstract boolean hasBeenStopped();
 

http://git-wip-us.apache.org/repos/asf/flink/blob/46ad4058/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
b/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
index ccde5d8..dd32b0d 100644
--- a/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
+++ b/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
@@ -647,7 +647,7 @@ public class YARNSessionFIFOITCase extends YarnTestBase {
 
 		LOG.info("Shutting down cluster. All tests passed");
 		// shutdown cluster
-		yarnCluster.shutdown();
+		yarnCluster.shutdown(false);
 		LOG.info("Finished testJavaAPI()");
 	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/46ad4058/flink-yarn/src/main/java/org/apache/flink/yarn/FlinkYarnClient.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/FlinkYarnClient.java b/flink-yarn/src/main/java/org/apache/flink/yarn/FlinkYarnClient.java
index 118f4ad..f82f013 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/FlinkYarnClient.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/FlinkYarnClient.java
@@ -435,11 +435,11 @@ public class FlinkYarnClient extends AbstractFlinkYarnClient {
 					+ "There are currently only " + freeClusterMem.totalFreeMemory + "MB available." + NOTE_RSC);
 
 		}
-		if( taskManagerMemoryMb > freeClusterMem.containerLimit) {
+		if(taskManagerMemoryMb > freeClusterMem.containerLimit) {
 			LOG.warn("The requested amount of memory for the TaskManagers ("+taskManagerMemoryMb+"MB)
is more than "
 					+ "the largest possible YARN container: "+freeClusterMem.containerLimit + NOTE_RSC);
 		}
-		if( jobManagerMemoryMb > freeClusterMem.containerLimit) {
+		if(jobManagerMemoryMb > freeClusterMem.containerLimit) {
 			LOG.warn("The requested amount of memory for the JobManager (" + jobManagerMemoryMb +
"MB) is more than "
 					+ "the largest possible YARN container: " + freeClusterMem.containerLimit + NOTE_RSC);
 		}

http://git-wip-us.apache.org/repos/asf/flink/blob/46ad4058/flink-yarn/src/main/java/org/apache/flink/yarn/FlinkYarnCluster.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/FlinkYarnCluster.java b/flink-yarn/src/main/java/org/apache/flink/yarn/FlinkYarnCluster.java
index 6dd84d6..e408edb 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/FlinkYarnCluster.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/FlinkYarnCluster.java
@@ -297,12 +297,12 @@ public class FlinkYarnCluster extends AbstractFlinkYarnCluster {
 		}
 		ApplicationReport lastReport = pollingRunner.getLastReport();
 		if(lastReport == null) {
-			LOG.warn("FlinkYarnCluster.hasFailed() has been called on a cluster. that didn't receive
a status so far." +
+			LOG.warn("FlinkYarnCluster.hasFailed() has been called on a cluster that didn't receive
a status so far." +
 					"The system might be in an erroneous state");
 			return false;
 		} else {
 			YarnApplicationState appState = lastReport.getYarnApplicationState();
-			boolean status= (appState == YarnApplicationState.FAILED ||
+			boolean status = (appState == YarnApplicationState.FAILED ||
 					appState == YarnApplicationState.KILLED);
 			if(status) {
 				LOG.warn("YARN reported application state {}", appState);
@@ -381,12 +381,13 @@ public class FlinkYarnCluster extends AbstractFlinkYarnCluster {
 	// -------------------------- Shutdown handling ------------------------
 
 	private AtomicBoolean hasBeenShutDown = new AtomicBoolean(false);
-	@Override
-	public void shutdown() {
-		shutdownInternal(true);
-	}
 
-	private void shutdownInternal(boolean removeShutdownHook) {
+	/**
+	 * Shutdown the YARN cluster.
+	 * @param failApplication whether we should fail the YARN application (in case of errors
in Flink)
+	 */
+	@Override
+	public void shutdown(boolean failApplication) {
 		if(!isConnected) {
 			throw new IllegalStateException("The cluster has been connected to the ApplicationMaster.");
 		}
@@ -394,16 +395,25 @@ public class FlinkYarnCluster extends AbstractFlinkYarnCluster {
 		if(hasBeenShutDown.getAndSet(true)) {
 			return;
 		}
-		// the session is being stopped explicitly.
-		if(removeShutdownHook) {
+
+		try {
 			Runtime.getRuntime().removeShutdownHook(clientShutdownHook);
+		} catch (IllegalStateException e) {
+			// we are already in the shutdown hook
 		}
+
 		if(actorSystem != null){
 			LOG.info("Sending shutdown request to the Application Master");
 			if(applicationClient != ActorRef.noSender()) {
 				try {
+					FinalApplicationStatus finalStatus;
+					if (failApplication) {
+						finalStatus = FinalApplicationStatus.FAILED;
+					} else {
+						finalStatus = FinalApplicationStatus.SUCCEEDED;
+					}
 					Future<Object> response = Patterns.ask(applicationClient,
-							new Messages.StopYarnSession(FinalApplicationStatus.SUCCEEDED,
+							new Messages.StopYarnSession(finalStatus,
 									"Flink YARN Client requested shutdown"),
 							new Timeout(akkaDuration));
 
@@ -457,7 +467,7 @@ public class FlinkYarnCluster extends AbstractFlinkYarnCluster {
 		@Override
 		public void run() {
 			LOG.info("Shutting down FlinkYarnCluster from the client shutdown hook");
-			shutdownInternal(false);
+			shutdown(true);
 		}
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/46ad4058/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationClient.scala
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationClient.scala b/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationClient.scala
index d6760ec..ec980d0 100644
--- a/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationClient.scala
+++ b/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationClient.scala
@@ -137,7 +137,7 @@ class ApplicationClient(flinkConfig: Configuration)
     case LocalGetYarnClusterStatus =>
       sender() ! latestClusterStatus
 
-      // Forward message to Application Master
+    // Forward message to Application Master
     case msg: StopAMAfterJob =>
       yarnJobManager foreach {
         _ forward msg

http://git-wip-us.apache.org/repos/asf/flink/blob/46ad4058/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMasterActor.scala
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMasterActor.scala
b/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMasterActor.scala
index 999610f..411808b 100644
--- a/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMasterActor.scala
+++ b/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMasterActor.scala
@@ -26,6 +26,7 @@ import akka.actor.ActorRef
 import org.apache.flink.api.common.JobID
 import org.apache.flink.configuration.ConfigConstants
 import org.apache.flink.runtime.ActorLogMessages
+import org.apache.flink.runtime.jobgraph.JobStatus
 import org.apache.flink.runtime.jobmanager.JobManager
 import org.apache.flink.runtime.messages.JobManagerMessages.{CurrentJobStatus, JobNotFound,
RequestJobStatus}
 import org.apache.flink.runtime.messages.Messages.Acknowledge
@@ -171,8 +172,13 @@ trait ApplicationMasterActor extends ActorLogMessages {
           if(jobStatus.status.isTerminalState) {
             log.info(s"Job with ID ${jobStatus.jobID} is in terminal state ${jobStatus.status}.
" +
               s"Shutting down YARN session")
-            self ! StopYarnSession(FinalApplicationStatus.SUCCEEDED,
-              s"The monitored job with ID ${jobStatus.jobID} has finished.")
+            if (jobStatus.status == JobStatus.FINISHED) {
+              self ! StopYarnSession(FinalApplicationStatus.SUCCEEDED,
+                s"The monitored job with ID ${jobStatus.jobID} has finished.")
+            } else {
+              self ! StopYarnSession(FinalApplicationStatus.FAILED,
+                s"The monitored job with ID ${jobStatus.jobID} has failed to complete.")
+            }
           } else {
             log.debug(s"Monitored job with ID ${jobStatus.jobID} is in state ${jobStatus.status}")
           }


Mime
View raw message