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/05/27 09:57:38 UTC
[1/3] flink git commit: [jobmanager] Fix potential null pointer
exception in jobmanager webfrontend
Repository: flink
Updated Branches:
refs/heads/master 3586ced35 -> 3c037be7d
[jobmanager] Fix potential null pointer exception in jobmanager webfrontend
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b2b0fe79
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b2b0fe79
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b2b0fe79
Branch: refs/heads/master
Commit: b2b0fe79d28ba8340028dad0b03409fcccc69a50
Parents: 3586ced
Author: Robert Metzger <rm...@apache.org>
Authored: Tue May 19 19:38:02 2015 +0200
Committer: Robert Metzger <rm...@apache.org>
Committed: Wed May 27 09:56:53 2015 +0200
----------------------------------------------------------------------
.../flink/api/common/ExecutionConfig.java | 2 +-
.../jobmanager/web/JobManagerInfoServlet.java | 28 +++++++++++++-------
2 files changed, 20 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b2b0fe79/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java b/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
index 8baedb4..04a518e 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/ExecutionConfig.java
@@ -605,7 +605,7 @@ public class ExecutionConfig implements Serializable {
}
/**
- * Interface for custom user configuration object registered at the execution config.
+ * Abstract class for a custom user configuration object registered at the execution config.
*
* This user config is accessible at runtime through
* getRuntimeContext().getExecutionConfig().getUserConfig()
http://git-wip-us.apache.org/repos/asf/flink/blob/b2b0fe79/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/web/JobManagerInfoServlet.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/web/JobManagerInfoServlet.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/web/JobManagerInfoServlet.java
index a407d95..6d58306 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/web/JobManagerInfoServlet.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/web/JobManagerInfoServlet.java
@@ -409,19 +409,29 @@ public class JobManagerInfoServlet extends HttpServlet {
wrt.write("\"Job parallelism\": \""+ec.getParallelism()+"\",");
wrt.write("\"Object reuse mode\": \""+ec.isObjectReuseEnabled()+"\"");
ExecutionConfig.GlobalJobParameters uc = ec.getGlobalJobParameters();
- Map<String, String> ucVals = uc.toMap();
- if(ucVals != null) {
- String ucString = "{";
- int i = 0;
- for (Map.Entry<String, String> ucVal: ucVals.entrySet()) {
- ucString += "\""+ucVal.getKey()+"\":\""+ucVal.getValue()+"\"";
- if (++i < ucVals.size()) {
- ucString += ",\n";
+ if(uc != null) {
+ Map<String, String> ucVals = uc.toMap();
+ if (ucVals != null) {
+ String ucString = "{";
+ int i = 0;
+ for (Map.Entry<String, String> ucVal : ucVals.entrySet()) {
+ ucString += "\"" + ucVal.getKey() + "\":\"" + ucVal.getValue() + "\"";
+ if (++i < ucVals.size()) {
+ ucString += ",\n";
+ }
}
+ wrt.write(", \"userConfig\": " + ucString + "}");
+ }
+ else {
+ LOG.info("GlobalJobParameters.toMap() did not return anything");
}
- wrt.write(", \"userConfig\": "+ucString+"}");
+ }
+ else {
+ LOG.info("No GlobalJobParameters were set in the execution config");
}
wrt.write("},");
+ } else {
+ LOG.warn("Unable to retrieve execution config from execution graph");
}
// write accumulators
[3/3] flink git commit: [yarn] Adjust default values for YARN heap
memory cutoff
Posted by rm...@apache.org.
[yarn] Adjust default values for YARN heap memory cutoff
This closes #717
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3c037be7
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3c037be7
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3c037be7
Branch: refs/heads/master
Commit: 3c037be7d2884ff0276952b30638852a01ba687b
Parents: 11b021b
Author: Robert Metzger <rm...@apache.org>
Authored: Fri May 22 18:10:15 2015 +0200
Committer: Robert Metzger <rm...@apache.org>
Committed: Wed May 27 09:56:54 2015 +0200
----------------------------------------------------------------------
.../flink/configuration/ConfigConstants.java | 12 ++++++++--
.../YARNSessionCapacitySchedulerITCase.java | 4 ++--
.../flink/yarn/YARNSessionFIFOITCase.java | 24 ++++++++++----------
.../org/apache/flink/yarn/FlinkYarnClient.java | 8 +++----
4 files changed, 28 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3c037be7/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
index 92acd3f..9d8da21 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
@@ -553,9 +553,17 @@ public final class ConfigConstants {
// ------------------------ YARN Configuration ------------------------
- public static final int DEFAULT_YARN_MIN_HEAP_CUTOFF = 384;
+ /**
+ * Minimum amount of Heap memory to subtract from the requested TaskManager size.
+ * We came up with these values experimentally.
+ * Flink fails when the cutoff is set only to 500 mb.
+ */
+ public static final int DEFAULT_YARN_MIN_HEAP_CUTOFF = 600;
- public static final float DEFAULT_YARN_HEAP_CUTOFF_RATIO = 0.15f;
+ /**
+ * Relative amount of memory to subtract from the requested memory.
+ */
+ public static final float DEFAULT_YARN_HEAP_CUTOFF_RATIO = 0.25f;
// ------------------------ File System Behavior ------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3c037be7/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java b/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
index 9c6748a..4051ff5 100644
--- a/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
+++ b/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YARNSessionCapacitySchedulerITCase.java
@@ -56,7 +56,7 @@ public class YARNSessionCapacitySchedulerITCase extends YarnTestBase {
LOG.info("Starting testClientStartup()");
runWithArgs(new String[]{"-j", flinkUberjar.getAbsolutePath(),
"-n", "1",
- "-jm", "512",
+ "-jm", "768",
"-tm", "1024", "-qu", "qa-team"},
"Number of connected TaskManagers changed to 1. Slots available: 1", null, RunTypes.YARN_SESSION, 0);
LOG.info("Finished testClientStartup()");
@@ -73,7 +73,7 @@ public class YARNSessionCapacitySchedulerITCase extends YarnTestBase {
addTestAppender(FlinkYarnClient.class, Level.WARN);
runWithArgs(new String[]{"-j", flinkUberjar.getAbsolutePath(),
"-n", "1",
- "-jm", "512",
+ "-jm", "768",
"-tm", "1024",
"-qu", "doesntExist"}, "to unknown queue: doesntExist", null, RunTypes.YARN_SESSION, 1);
checkForLogString("The specified queue 'doesntExist' does not exist. Available queues: default, qa-team");
http://git-wip-us.apache.org/repos/asf/flink/blob/3c037be7/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 e22e0ef..f4febed 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
@@ -102,7 +102,7 @@ public class YARNSessionFIFOITCase extends YarnTestBase {
LOG.info("Starting testClientStartup()");
runWithArgs(new String[]{"-j", flinkUberjar.getAbsolutePath(),
"-n", "1",
- "-jm", "512",
+ "-jm", "768",
"-tm", "1024",
"-s", "2" // Test that 2 slots are started on the TaskManager.
},
@@ -119,7 +119,7 @@ public class YARNSessionFIFOITCase extends YarnTestBase {
addTestAppender(FlinkYarnSessionCli.class, Level.INFO);
Runner runner = startWithArgs(new String[]{"-j", flinkUberjar.getAbsolutePath(),
"-n", "1",
- "-jm", "512",
+ "-jm", "768",
"-tm", "1024",
"--detached"},
"Flink JobManager is now running on", RunTypes.YARN_SESSION);
@@ -164,7 +164,7 @@ public class YARNSessionFIFOITCase extends YarnTestBase {
LOG.info("Starting testTaskManagerFailure()");
Runner runner = startWithArgs(new String[]{"-j", flinkUberjar.getAbsolutePath(),
"-n", "1",
- "-jm", "512",
+ "-jm", "768",
"-tm", "1024",
"-Dfancy-configuration-value=veryFancy",
"-Dyarn.maximum-failed-containers=3"},
@@ -332,7 +332,7 @@ public class YARNSessionFIFOITCase extends YarnTestBase {
LOG.info("Starting testNonexistingQueue()");
runWithArgs(new String[]{"-j", flinkUberjar.getAbsolutePath(),
"-n", "1",
- "-jm", "512",
+ "-jm", "768",
"-tm", "1024",
"-qu", "doesntExist"}, "Number of connected TaskManagers changed to 1. Slots available: 1", null, RunTypes.YARN_SESSION, 0);
LOG.info("Finished testNonexistingQueue()");
@@ -408,7 +408,7 @@ public class YARNSessionFIFOITCase extends YarnTestBase {
"-yj", flinkUberjar.getAbsolutePath(),
"-yn", "1",
"-ys", "2", //test that the job is executed with a DOP of 2
- "-yjm", "512",
+ "-yjm", "768",
"-ytm", "1024", exampleJarLocation.getAbsolutePath()},
/* test succeeded after this string */
"Job execution switched to status FINISHED.",
@@ -431,7 +431,7 @@ public class YARNSessionFIFOITCase extends YarnTestBase {
"-m", "yarn-cluster",
"-yj", flinkUberjar.getAbsolutePath(),
"-yn", "1",
- "-yjm", "512",
+ "-yjm", "768",
"-ytm", "1024", exampleJarLocation.getAbsolutePath()},
/* test succeeded after this string */
"Job execution switched to status FINISHED.",
@@ -467,7 +467,7 @@ public class YARNSessionFIFOITCase extends YarnTestBase {
Runner runner = startWithArgs(new String[]{"run", "-m", "yarn-cluster", "-yj", flinkUberjar.getAbsolutePath(),
"-yn", "1",
- "-yjm", "512",
+ "-yjm", "768",
"-yD", "yarn.heap-cutoff-ratio=0.5", // test if the cutoff is passed correctly
"-ytm", "1024",
"-ys", "2", // test requesting slots from YARN.
@@ -541,8 +541,8 @@ public class YARNSessionFIFOITCase extends YarnTestBase {
Assert.assertNotNull("Unable to locate JobManager log", jobmanagerLog);
content = FileUtils.readFileToString(jobmanagerLog);
// expecting 512 mb, because TM was started with 1024, we cut off 50% (NOT THE DEFAULT VALUE).
- Assert.assertTrue("Expected string 'Starting TM with command=$JAVA_HOME/bin/java -Xms512m -Xmx512m' not found in JobManager log: '"+jobmanagerLog+"'",
- content.contains("Starting TM with command=$JAVA_HOME/bin/java -Xms512m -Xmx512m"));
+ Assert.assertTrue("Expected string 'Starting TM with command=$JAVA_HOME/bin/java -Xms424m -Xmx424m' not found in JobManager log: '"+jobmanagerLog+"'",
+ content.contains("Starting TM with command=$JAVA_HOME/bin/java -Xms424m -Xmx424m"));
Assert.assertTrue("Expected string ' (2/2) (attempt #0) to ' not found in JobManager log." +
"This string checks that the job has been started with a parallelism of 2. Log contents: '"+jobmanagerLog+"'",
content.contains(" (2/2) (attempt #0) to "));
@@ -558,7 +558,7 @@ public class YARNSessionFIFOITCase extends YarnTestBase {
} catch(Throwable t) {
LOG.warn("Error while detached yarn session was running", t);
- Assert.fail();
+ Assert.fail(t.getMessage());
}
}
@@ -604,8 +604,8 @@ public class YARNSessionFIFOITCase extends YarnTestBase {
AbstractFlinkYarnClient flinkYarnClient = FlinkYarnSessionCli.getFlinkYarnClient();
Assert.assertNotNull("unable to get yarn client", flinkYarnClient);
flinkYarnClient.setTaskManagerCount(1);
- flinkYarnClient.setJobManagerMemory(512);
- flinkYarnClient.setTaskManagerMemory(512);
+ flinkYarnClient.setJobManagerMemory(768);
+ flinkYarnClient.setTaskManagerMemory(768);
flinkYarnClient.setLocalJarPath(new Path(flinkUberjar.getAbsolutePath()));
String confDirPath = System.getenv("FLINK_CONF_DIR");
flinkYarnClient.setConfigurationDirectory(confDirPath);
http://git-wip-us.apache.org/repos/asf/flink/blob/3c037be7/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 502d72d..88b2987 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
@@ -99,8 +99,8 @@ public class FlinkYarnClient extends AbstractFlinkYarnClient {
/**
* Minimum memory requirements, checked by the Client.
*/
- private static final int MIN_JM_MEMORY = 128;
- private static final int MIN_TM_MEMORY = 128;
+ private static final int MIN_JM_MEMORY = 768; // the minimum memory should be higher than the min heap cutoff
+ private static final int MIN_TM_MEMORY = 768;
private Configuration conf;
private YarnClient yarnClient;
@@ -164,7 +164,7 @@ public class FlinkYarnClient extends AbstractFlinkYarnClient {
@Override
public void setJobManagerMemory(int memoryMb) {
if(memoryMb < MIN_JM_MEMORY) {
- throw new IllegalArgumentException("The JobManager memory is below the minimum required memory amount "
+ throw new IllegalArgumentException("The JobManager memory ("+memoryMb+") is below the minimum required memory amount "
+ "of "+MIN_JM_MEMORY+" MB");
}
this.jobManagerMemoryMb = memoryMb;
@@ -173,7 +173,7 @@ public class FlinkYarnClient extends AbstractFlinkYarnClient {
@Override
public void setTaskManagerMemory(int memoryMb) {
if(memoryMb < MIN_TM_MEMORY) {
- throw new IllegalArgumentException("The TaskManager memory is below the minimum required memory amount "
+ throw new IllegalArgumentException("The TaskManager memory ("+memoryMb+") is below the minimum required memory amount "
+ "of "+MIN_TM_MEMORY+" MB");
}
this.taskManagerMemoryMb = memoryMb;
[2/3] flink git commit: [FLINK-2079] Add TaskManager deathwatch
thread for YARN case
Posted by rm...@apache.org.
[FLINK-2079] Add TaskManager deathwatch thread for YARN case
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/11b021b0
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/11b021b0
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/11b021b0
Branch: refs/heads/master
Commit: 11b021b0fb36503c06596323b39d531225057f1e
Parents: b2b0fe7
Author: Robert Metzger <rm...@apache.org>
Authored: Fri May 22 13:51:02 2015 +0200
Committer: Robert Metzger <rm...@apache.org>
Committed: Wed May 27 09:56:54 2015 +0200
----------------------------------------------------------------------
.../org/apache/flink/configuration/ConfigConstants.java | 5 +++++
.../org/apache/flink/runtime/process/ProcessReaper.java | 2 +-
.../scala/org/apache/flink/runtime/akka/AkkaUtils.scala | 9 ++++++++-
.../org/apache/flink/runtime/taskmanager/TaskManager.scala | 5 ++---
.../apache/flink/yarn/appMaster/YarnTaskManagerRunner.java | 3 +++
5 files changed, 19 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/11b021b0/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
index 26d4fbe..92acd3f 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
@@ -406,6 +406,11 @@ public final class ConfigConstants {
* Timeout for all blocking calls that look up remote actors
*/
public static final String AKKA_LOOKUP_TIMEOUT = "akka.lookup.timeout";
+
+ /**
+ * Exit JVM on fatal Akka errors
+ */
+ public static final String AKKA_JVM_EXIT_ON_FATAL_ERROR = "akka.jvm-exit-on-fatal-error";
// ----------------------------- Streaming --------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/11b021b0/flink-runtime/src/main/java/org/apache/flink/runtime/process/ProcessReaper.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/process/ProcessReaper.java b/flink-runtime/src/main/java/org/apache/flink/runtime/process/ProcessReaper.java
index 644d7b7..09e1839 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/process/ProcessReaper.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/process/ProcessReaper.java
@@ -56,7 +56,7 @@ public class ProcessReaper extends UntypedActor {
Thread.sleep(100);
}
catch (InterruptedException e) {
- // not really problem if we don't sleep...
+ // not really a problem if we don't sleep...
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/11b021b0/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala
index 5b33017..7ffaddd 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala
@@ -139,6 +139,13 @@ object AkkaUtils {
val lifecycleEvents = configuration.getBoolean(ConfigConstants.AKKA_LOG_LIFECYCLE_EVENTS,
ConfigConstants.DEFAULT_AKKA_LOG_LIFECYCLE_EVENTS)
+ val jvmExitOnFatalError = if (
+ configuration.getBoolean(ConfigConstants.AKKA_JVM_EXIT_ON_FATAL_ERROR, false)){
+ "on"
+ } else {
+ "off"
+ }
+
val logLifecycleEvents = if (lifecycleEvents) "on" else "off"
val logLevel = getLogLevel
@@ -152,7 +159,7 @@ object AkkaUtils {
| logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"
| log-config-on-start = off
|
- | jvm-exit-on-fatal-error = off
+ | jvm-exit-on-fatal-error = $jvmExitOnFatalError
|
| serialize-messages = off
|
http://git-wip-us.apache.org/repos/asf/flink/blob/11b021b0/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 8a45fa4..7bf5bc5 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
@@ -158,7 +158,6 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
private var heartbeatScheduler: Option[Cancellable] = None
-
// --------------------------------------------------------------------------
// Actor messages and life cycle
// --------------------------------------------------------------------------
@@ -192,7 +191,7 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
*/
override def postStop(): Unit = {
log.info(s"Stopping TaskManager ${self.path.toSerializationFormat}.")
-
+
cancelAndClearEverything(new Exception("TaskManager is shutting down."))
if (isConnected) {
@@ -1289,7 +1288,7 @@ object TaskManager {
streamingMode,
taskManagerClass)
- // start a process reaper that watches the JobManager. If the JobManager actor dies,
+ // start a process reaper that watches the JobManager. If the TaskManager actor dies,
// the process reaper will kill the JVM process (to ensure easy failure detection)
LOG.debug("Starting TaskManager process reaper")
taskManagerSystem.actorOf(
http://git-wip-us.apache.org/repos/asf/flink/blob/11b021b0/flink-yarn/src/main/java/org/apache/flink/yarn/appMaster/YarnTaskManagerRunner.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/appMaster/YarnTaskManagerRunner.java b/flink-yarn/src/main/java/org/apache/flink/yarn/appMaster/YarnTaskManagerRunner.java
index 564a0bd..3f13990 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/appMaster/YarnTaskManagerRunner.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/appMaster/YarnTaskManagerRunner.java
@@ -85,6 +85,9 @@ public class YarnTaskManagerRunner {
LOG.info("YARN daemon runs as '" + UserGroupInformation.getCurrentUser().getShortUserName()
+"' setting user to execute Flink TaskManager to '"+yarnClientUsername+"'");
+ // tell akka to die in case of an error
+ configuration.setBoolean(ConfigConstants.AKKA_JVM_EXIT_ON_FATAL_ERROR, true);
+
UserGroupInformation ugi = UserGroupInformation.createRemoteUser(yarnClientUsername);
for (Token<? extends TokenIdentifier> toks : UserGroupInformation.getCurrentUser().getTokens()) {
ugi.addToken(toks);