[runtime] Avoid IllegalStateException when TaskManager is killed during startup

This closes #587

flink-runtime/src/main/java/org/apache/flink/runtime/io/disk/iomanager/IOManager.java

@@ -103,7 +103,14 @@ public abstract class IOManager {
 				shutdown();
 			}
 		};
-		Runtime.getRuntime().addShutdownHook(this.shutdownHook);
+		try {
+			Runtime.getRuntime().addShutdownHook(this.shutdownHook);
+		} catch (IllegalStateException e) {
+			// race, JVM is in shutdown already, we can safely ignore this
+			LOG.debug("Unable to add shutdown hook, shutdown already in progress", e);
+		} catch (Throwable t) {
+			LOG.warn("Error while adding shutdown hook for IOManager", t);
+		}
 	}
 
 	/**
@@ -132,6 +139,7 @@ public abstract class IOManager {
 			}
 			catch (IllegalStateException e) {
 				// race, JVM is in shutdown already, we can safely ignore this
+				LOG.debug("Unable to remove shutdown hook, shutdown already in progress", e);
 			}
 			catch (Throwable t) {
 				LOG.warn("Exception while unregistering IOManager's shutdown hook.", t);

flink-yarn-tests/src/main/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java

@@ -123,7 +123,13 @@ public class YARNSessionFIFOITCase extends YarnTestBase {
 		checkForLogString("The Flink YARN client has been started in detached mode");
 
 		Assert.assertFalse("The runner should detach.", runner.isAlive());
-		sleep(5000); // wait for 5 seconds to make sure the the app has been started completely before we kill it
+
+		LOG.info("Waiting until two containers are running");
+		// wait until two containers are running
+		while(getRunningContainers() < 2) {
+			sleep(500);
+		}
+		LOG.info("Two containers are running. Killing the application");
 
 		// kill application "externally".
 		try {