[FLINK-5830] [distributed coordination] Handle OutOfMemory and fatal …

…errors during process async message in akka rpc actor

This closes apache#3360

flink-core/src/main/java/org/apache/flink/util/ExceptionUtils.java

@@ -112,6 +112,19 @@ public static void rethrowIfFatalError(Throwable t) {
 		}
 	}
 
+	/**
+	 * Rethrows the given {@code Throwable}, if it represents an error that is fatal to the JVM
+	 * or an out-of-memory error. See {@link ExceptionUtils#isJvmFatalError(Throwable)} for a
+	 * definition of fatal errors.
+	 *
+	 * @param t The Throwable to check and rethrow.
+	 */
+	public static void rethrowIfFatalErrorOrOOM(Throwable t) {
+		if (isJvmFatalError(t) || t instanceof OutOfMemoryError) {
+			throw (Error) t;
+		}
+	}
+
 	/**
 	 * Adds a new exception as a {@link Throwable#addSuppressed(Throwable) suppressed exception}
 	 * to a prior exception, or returns the new exception, if no prior exception exists.

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcActor.java

@@ -38,6 +38,7 @@
 import org.apache.flink.runtime.rpc.akka.messages.RunAsync;
 
 import org.apache.flink.runtime.rpc.exceptions.RpcConnectionException;
+import org.apache.flink.util.ExceptionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -271,8 +272,9 @@ else if (runAsync.getDelay() == 0) {
 			// run immediately
 			try {
 				runAsync.getRunnable().run();
-			} catch (final Throwable e) {
-				LOG.error("Caught exception while executing runnable in main thread.", e);
+			} catch (Throwable t) {
+				LOG.error("Caught exception while executing runnable in main thread.", t);
+				ExceptionUtils.rethrowIfFatalErrorOrOOM(t);
 			}
 		}
 		else {