[FLINK-19022] Let TaskExecutor, Dispatcher, RM fail fatally if onStar…

…t throws a Throwable

If TaskExecutor., Dispatcher. or ResourceManager.onStart throws a Throwable, the system is
no longer in a valid state. Hence, this commit ensures that the components will properly
call FatalErrorHandler.onFatalError in such a case.

flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java

@@ -202,8 +202,8 @@ public CompletableFuture<ApplicationStatus> getShutDownFuture() {
 	public void onStart() throws Exception {
 		try {
 			startDispatcherServices();
-		} catch (Exception e) {
-			final DispatcherException exception = new DispatcherException(String.format("Could not start the Dispatcher %s", getAddress()), e);
+		} catch (Throwable t) {
+			final DispatcherException exception = new DispatcherException(String.format("Could not start the Dispatcher %s", getAddress()), t);
 			onFatalError(exception);
 			throw exception;
 		}

flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java

@@ -216,8 +216,8 @@ public ResourceManager(
 	public final void onStart() throws Exception {
 		try {
 			startResourceManagerServices();
-		} catch (Exception e) {
-			final ResourceManagerException exception = new ResourceManagerException(String.format("Could not start the ResourceManager %s", getAddress()), e);
+		} catch (Throwable t) {
+			final ResourceManagerException exception = new ResourceManagerException(String.format("Could not start the ResourceManager %s", getAddress()), t);
 			onFatalError(exception);
 			throw exception;
 		}

flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java

@@ -360,8 +360,8 @@ public CompletableFuture<Collection<LogInfo>> requestLogList(Time timeout) {
 	public void onStart() throws Exception {
 		try {
 			startTaskExecutorServices();
-		} catch (Exception e) {
-			final TaskManagerException exception = new TaskManagerException(String.format("Could not start the TaskExecutor %s", getAddress()), e);
+		} catch (Throwable t) {
+			final TaskManagerException exception = new TaskManagerException(String.format("Could not start the TaskExecutor %s", getAddress()), t);
 			onFatalError(exception);
 			throw exception;
 		}