[FLINK-3134][yarn] asynchronous YarnJobManager heartbeats

- use AMRMClientAsync instead of AMRMClient
- handle allocation and startup of containers in callbacks
- remove YarnHeartbeat message

The AMRMClientAsync uses one thread to communicate with the resource
manager and an additional thread to execute the callbacks.

This closes apache#1450.

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

@@ -284,14 +284,14 @@ public void testTaskManagerFailure() {
  }
 
  // stateful termination check:
- // wait until we saw a container being killed and AFTERWARDS a new one launced
+ // wait until we saw a container being killed and AFTERWARDS a new one launched
  boolean ok = false;
  do {
  LOG.debug("Waiting for correct order of events. Output: {}", errContent.toString());
 
  String o = errContent.toString();
  int killedOff = o.indexOf("Container killed by the ApplicationMaster");
- if(killedOff != -1) {
+ if (killedOff != -1) {
  o = o.substring(killedOff);
  ok = o.indexOf("Launching container") > 0;
  }

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

@@ -89,7 +89,10 @@ public abstract class YarnTestBase extends TestLogger {
 
  /** These strings are white-listed, overriding teh prohibited strings */
  protected final static String[] WHITELISTED_STRINGS = {
- "akka.remote.RemoteTransportExceptionNoStackTrace"
+ "akka.remote.RemoteTransportExceptionNoStackTrace",
+ // workaround for annoying InterruptedException logging:
+ // https://issues.apache.org/jira/browse/YARN-1022
+ "java.lang.InterruptedException"
  };
 
  // Temp directory which is deleted after the unit test.

flink-yarn-tests/src/main/scala/org/apache/flink/yarn/TestingYarnJobManager.scala

@@ -47,7 +47,6 @@ import scala.concurrent.duration.FiniteDuration
  * @param defaultExecutionRetries Number of default execution retries
  * @param delayBetweenRetries Delay between retries
  * @param timeout Timeout for futures
- * @param mode StreamingMode in which the system shall be started
  * @param leaderElectionService LeaderElectionService to participate in the leader election
  */
 class TestingYarnJobManager(

flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationClient.scala

@@ -21,6 +21,8 @@ package org.apache.flink.yarn
 import java.util.UUID
 
 import akka.actor._
+import akka.pattern
+import akka.util.Timeout
 import grizzled.slf4j.Logger
 import org.apache.flink.configuration.Configuration
 import org.apache.flink.runtime.leaderretrieval.{LeaderRetrievalListener, LeaderRetrievalService}