flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From m..@apache.org
Subject flink git commit: [FLINK-3134][yarn] asynchronous YarnJobManager heartbeats
Date Thu, 17 Dec 2015 14:20:25 GMT
Repository: flink
Updated Branches:
  refs/heads/master d6e9fa046 -> 4e52fe430


[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 #1450.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4e52fe43
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4e52fe43
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4e52fe43

Branch: refs/heads/master
Commit: 4e52fe4304566e5239996b3d48290e0c1f0772e8
Parents: d6e9fa0
Author: Maximilian Michels <mxm@apache.org>
Authored: Mon Dec 7 14:35:04 2015 +0100
Committer: Maximilian Michels <mxm@apache.org>
Committed: Thu Dec 17 15:19:48 2015 +0100

----------------------------------------------------------------------
 .../flink/yarn/YARNSessionFIFOITCase.java       |   4 +-
 .../org/apache/flink/yarn/YarnTestBase.java     |   5 +-
 .../flink/yarn/TestingYarnJobManager.scala      |   1 -
 .../apache/flink/yarn/ApplicationClient.scala   |   2 +
 .../org/apache/flink/yarn/YarnJobManager.scala  | 560 ++++++++++---------
 5 files changed, 301 insertions(+), 271 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/4e52fe43/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 69925f2..14ee8ec 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
@@ -284,14 +284,14 @@ public class YARNSessionFIFOITCase extends YarnTestBase {
 		}
 
 		// 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;
 			}

http://git-wip-us.apache.org/repos/asf/flink/blob/4e52fe43/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YarnTestBase.java
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YarnTestBase.java b/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YarnTestBase.java
index 5ce528e..d3132d7 100644
--- a/flink-yarn-tests/src/main/java/org/apache/flink/yarn/YarnTestBase.java
+++ b/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.

http://git-wip-us.apache.org/repos/asf/flink/blob/4e52fe43/flink-yarn-tests/src/main/scala/org/apache/flink/yarn/TestingYarnJobManager.scala
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/src/main/scala/org/apache/flink/yarn/TestingYarnJobManager.scala
b/flink-yarn-tests/src/main/scala/org/apache/flink/yarn/TestingYarnJobManager.scala
index 3379325..a7f21e5 100644
--- a/flink-yarn-tests/src/main/scala/org/apache/flink/yarn/TestingYarnJobManager.scala
+++ b/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(

http://git-wip-us.apache.org/repos/asf/flink/blob/4e52fe43/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationClient.scala
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationClient.scala b/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationClient.scala
index 79717ef..8b2c06f 100644
--- a/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationClient.scala
+++ b/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}

http://git-wip-us.apache.org/repos/asf/flink/blob/4e52fe43/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnJobManager.scala
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnJobManager.scala b/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnJobManager.scala
index 8494f08..735f567 100644
--- a/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnJobManager.scala
+++ b/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnJobManager.scala
@@ -48,7 +48,8 @@ import org.apache.hadoop.yarn.api.ApplicationConstants
 import org.apache.hadoop.yarn.api.ApplicationConstants.Environment
 import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse
 import org.apache.hadoop.yarn.api.records._
-import org.apache.hadoop.yarn.client.api.{NMClient, AMRMClient}
+import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync
+import org.apache.hadoop.yarn.client.api.NMClient
 import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest
 import org.apache.hadoop.yarn.conf.YarnConfiguration
 import org.apache.hadoop.yarn.exceptions.YarnException
@@ -108,7 +109,7 @@ class YarnJobManager(
   val FAST_YARN_HEARTBEAT_DELAY: FiniteDuration = 500 milliseconds
   val DEFAULT_YARN_HEARTBEAT_DELAY: FiniteDuration = 5 seconds
   val YARN_HEARTBEAT_DELAY: FiniteDuration =
-    if(flinkConfiguration.getString(ConfigConstants.YARN_HEARTBEAT_DELAY_SECONDS, null) ==
null) {
+    if (flinkConfiguration.getString(ConfigConstants.YARN_HEARTBEAT_DELAY_SECONDS, null)
== null) {
       DEFAULT_YARN_HEARTBEAT_DELAY
     } else {
       FiniteDuration(
@@ -123,9 +124,9 @@ class YarnJobManager(
   val detached = java.lang.Boolean.valueOf(env.get(FlinkYarnClientBase.ENV_DETACHED))
   var stopWhenJobFinished: JobID = null
 
-  var rmClientOption: Option[AMRMClient[ContainerRequest]] = None
+  var rmClientOption: Option[AMRMClientAsync[ContainerRequest]] = None
   var nmClientOption: Option[NMClient] = None
-  var messageListener:Option[ActorRef] = None
+  var messageListener: Option[ActorRef] = None
   var containerLaunchContext: Option[ContainerLaunchContext] = None
 
   var runningContainers = 0 // number of currently running containers
@@ -155,27 +156,28 @@ class YarnJobManager(
 
       rmClientOption foreach {
         rmClient =>
-          Try(rmClient.unregisterApplicationMaster(status, diag, "")).recover{
+          Try(rmClient.unregisterApplicationMaster(status, diag, "")).recover {
             case t: Throwable => log.error("Could not unregister the application master.",
t)
           }
 
-          Try(rmClient.close()).recover{
-            case t:Throwable => log.error("Could not close the AMRMClient.", t)
+          Try(rmClient.stop()).recover {
+            case t: Throwable => log.error("Could not close the AMRMClient.", t)
           }
+
       }
 
       rmClientOption = None
 
       nmClientOption foreach {
         nmClient =>
-        Try(nmClient.close()).recover{
-          case t: Throwable => log.error("Could not close the NMClient.", t)
-        }
+          Try(nmClient.close()).recover {
+            case t: Throwable => log.error("Could not close the NMClient.", t)
+          }
       }
 
       nmClientOption = None
       messageListener foreach {
-          _ ! decorateMessage(JobManagerStopped)
+        _ ! decorateMessage(JobManagerStopped)
       }
 
       context.system.shutdown()
@@ -194,6 +196,17 @@ class YarnJobManager(
       val jobId = msg.jobId
       log.info(s"ApplicatonMaster will shut down YARN session when job $jobId has finished.")
       stopWhenJobFinished = jobId
+      // trigger regular job status messages (if this is a per-job yarn cluster)
+      if (stopWhenJobFinished != null) {
+        context.system.scheduler.schedule(0 seconds,
+          YARN_HEARTBEAT_DELAY,
+          new Runnable {
+            override def run(): Unit = {
+              self ! decorateMessage(RequestJobStatus(stopWhenJobFinished))
+            }
+          }
+        )
+      }
       sender() ! decorateMessage(Acknowledge)
 
 
@@ -209,19 +222,19 @@ class YarnJobManager(
 
     case jnf: JobNotFound =>
       log.warn(s"Job with ID ${jnf.jobID} not found in JobManager")
-      if(stopWhenJobFinished == null) {
+      if (stopWhenJobFinished == null) {
         log.warn("The ApplicationMaster didn't expect to receive this message")
       }
 
     case jobStatus: CurrentJobStatus =>
-      if(stopWhenJobFinished == null) {
+      if (stopWhenJobFinished == null) {
         log.warn(s"Received job status $jobStatus which wasn't requested.")
       } else {
-        if(stopWhenJobFinished != jobStatus.jobID) {
+        if (stopWhenJobFinished != jobStatus.jobID) {
           log.warn(s"Received job status for job ${jobStatus.jobID} but expected status for
" +
             s"job $stopWhenJobFinished")
         } else {
-          if(jobStatus.status.isTerminalState) {
+          if (jobStatus.status.isTerminalState) {
             log.info(s"Job with ID ${jobStatus.jobID} is in terminal state ${jobStatus.status}.
" +
               s"Shutting down YARN session")
             if (jobStatus.status == JobStatus.FINISHED) {
@@ -242,238 +255,18 @@ class YarnJobManager(
           }
         }
       }
-
-    case HeartbeatWithYarn =>
-      // piggyback on the YARN heartbeat to check if the job has finished
-      if(stopWhenJobFinished != null) {
-        self ! decorateMessage(RequestJobStatus(stopWhenJobFinished))
-      }
-      rmClientOption match {
-        case Some(rmClient) =>
-          log.debug("Send heartbeat to YARN")
-          val response = rmClient.allocate(runningContainers.toFloat / numTaskManagers)
-
-          // ---------------------------- handle YARN responses -------------
-
-          val newlyAllocatedContainers = response.getAllocatedContainers.asScala
-
-          newlyAllocatedContainers.foreach {
-            container => log.info(s"Got new container for allocation: ${container.getId}")
-          }
-
-          allocatedContainersList ++= newlyAllocatedContainers
-          numPendingRequests = math.max(0, numPendingRequests - newlyAllocatedContainers.length)
-
-          val completedContainerStatuses = response.getCompletedContainersStatuses.asScala
-          val idStatusMap = completedContainerStatuses
-            .map(status => (status.getContainerId, status)).toMap
-
-          completedContainerStatuses.foreach {
-            status => log.info(s"Container ${status.getContainerId} is completed " +
-              s"with diagnostics: ${status.getDiagnostics}")
-          }
-
-          // get failed containers (returned containers are also completed, so we have to
-          // distinguish if it was running before).
-          val (completedContainers, remainingRunningContainers) = runningContainersList
-            .partition(idStatusMap contains _.getId)
-
-          completedContainers.foreach {
-            container =>
-              val status = idStatusMap(container.getId)
-              failedContainers += 1
-              runningContainers -= 1
-              log.info(s"Container ${status.getContainerId} was a running container. " +
-                s"Total failed containers $failedContainers.")
-              val detail = status.getExitStatus match {
-                case -103 => "Vmem limit exceeded";
-                case -104 => "Pmem limit exceeded";
-                case _ => ""
-              }
-              messageListener foreach {
-                _ ! decorateMessage(
-                  YarnMessage(s"Diagnostics for containerID=${status.getContainerId} in "
+
-                    s"state=${status.getState}.\n${status.getDiagnostics} $detail")
-                )
-              }
-          }
-
-          runningContainersList = remainingRunningContainers
-
-          // return containers if the RM wants them and we haven't allocated them yet.
-          val preemptionMessage = response.getPreemptionMessage
-          if(preemptionMessage != null) {
-            log.info(s"Received preemtion message from YARN $preemptionMessage.")
-            val contract = preemptionMessage.getContract
-            if(contract != null) {
-              tryToReturnContainers(contract.getContainers.asScala.toSet)
-            }
-            val strictContract = preemptionMessage.getStrictContract
-            if(strictContract != null) {
-              tryToReturnContainers(strictContract.getContainers.asScala.toSet)
-            }
-          }
-
-          // ---------------------------- decide if we need to do anything ---------
-
-          // check if we want to start some of our allocated containers.
-          if(runningContainers < numTaskManagers) {
-            val missingContainers = numTaskManagers - runningContainers
-            log.info(s"The user requested $numTaskManagers containers, $runningContainers
" +
-              s"running. $missingContainers containers missing")
-
-            val numStartedContainers = startTMsInAllocatedContainers(missingContainers)
-
-            // if there are still containers missing, request them from YARN
-            val toAllocateFromYarn = Math.max(
-              missingContainers - numStartedContainers - numPendingRequests,
-              0)
-
-            if (toAllocateFromYarn > 0) {
-              val reallocate = flinkConfiguration
-                .getBoolean(ConfigConstants.YARN_REALLOCATE_FAILED_CONTAINERS, true)
-              log.info(s"There are $missingContainers containers missing." +
-                s" $numPendingRequests are already requested. " +
-                s"Requesting $toAllocateFromYarn additional container(s) from YARN. " +
-                s"Reallocation of failed containers is enabled=$reallocate " +
-                s"('${ConfigConstants.YARN_REALLOCATE_FAILED_CONTAINERS}')")
-              // there are still containers missing. Request them from YARN
-              if (reallocate) {
-                for(i <- 1 to toAllocateFromYarn) {
-                  val containerRequest = getContainerRequest(memoryPerTaskManager)
-                  rmClient.addContainerRequest(containerRequest)
-                  numPendingRequests += 1
-                  log.info("Requested additional container from YARN. Pending requests "
+
-                    s"$numPendingRequests.")
-                }
-              }
-            }
-          }
-
-          if(runningContainers >= numTaskManagers && allocatedContainersList.nonEmpty)
{
-            log.info(s"Flink has ${allocatedContainersList.size} allocated containers which
" +
-              s"are not needed right now. Returning them")
-            for(container <- allocatedContainersList) {
-              rmClient.releaseAssignedContainer(container.getId)
-            }
-            allocatedContainersList = List()
-          }
-
-          // maxFailedContainers == -1 is infinite number of retries.
-          if(maxFailedContainers != -1 && failedContainers >= maxFailedContainers)
{
-            val msg = s"Stopping YARN session because the number of failed " +
-              s"containers ($failedContainers) exceeded the maximum failed container " +
-              s"count ($maxFailedContainers). This number is controlled by " +
-              s"the '${ConfigConstants.YARN_MAX_FAILED_CONTAINERS}' configuration " +
-              s"setting. By default its the number of requested containers"
-            log.error(msg)
-            self ! decorateMessage(StopYarnSession(FinalApplicationStatus.FAILED, msg))
-          }
-
-          // schedule next heartbeat:
-          if (runningContainers < numTaskManagers) {
-            // we don't have the requested number of containers. Do fast polling
-            context.system.scheduler.scheduleOnce(
-              FAST_YARN_HEARTBEAT_DELAY,
-              self,
-              decorateMessage(HeartbeatWithYarn))
-          } else {
-            // everything is good, slow down polling
-            context.system.scheduler.scheduleOnce(
-              YARN_HEARTBEAT_DELAY,
-              self,
-              decorateMessage(HeartbeatWithYarn))
-          }
-        case None =>
-          log.error("The AMRMClient was not set.")
-          self ! decorateMessage(
-            StopYarnSession(
-              FinalApplicationStatus.FAILED,
-              "Fatal error in AM: AMRMClient was not set")
-          )
-      }
-      log.debug(s"Processed Heartbeat with RMClient. Running containers $runningContainers,
" +
-        s"failed containers $failedContainers, " +
-        s"allocated containers ${allocatedContainersList.size}.")
   }
 
-  /** Starts min(numTMsToStart, allocatedContainersList.size) TaskManager in the available
-    * allocated containers. The number of successfully started TaskManagers is returned.
-    *
-    * @param numTMsToStart Number of TaskManagers to start if enough allocated containers
are
-    *                      available. If not, then all allocated containers are used
-    * @return Number of successfully started TaskManagers
-    */
-  private def startTMsInAllocatedContainers(numTMsToStart: Int): Int = {
-    // not enough containers running
-    if (allocatedContainersList.nonEmpty) {
-      log.info(s"${allocatedContainersList.size} containers already allocated by YARN. "
+
-        "Starting...")
-
-      nmClientOption match {
-        case Some(nmClient) =>
-          containerLaunchContext match {
-            case Some(ctx) =>
-              val (containersToBeStarted, remainingContainers) = allocatedContainersList
-                .splitAt(numTMsToStart)
-
-              val startedContainers = containersToBeStarted.flatMap {
-                container =>
-                  try {
-                    nmClient.startContainer(container, ctx)
-                    val message = s"Launching container (${container.getId} on host " +
-                      s"${container.getNodeId.getHost})."
-                    log.info(message)
-
-                    messageListener foreach {
-                      _ ! decorateMessage(YarnMessage(message))
-                    }
-
-                    Some(container)
-                  } catch {
-                    case e: YarnException =>
-                      log.error(s"Exception while starting YARN " +
-                        s"container ${container.getId} on " +
-                        s"host ${container.getNodeId.getHost}", e)
-                      None
-                  }
-              }
-
-              runningContainers += startedContainers.length
-              runningContainersList :::= startedContainers
-
-              allocatedContainersList = remainingContainers
-
-              startedContainers.length
-            case None =>
-              log.error("The ContainerLaunchContext was not set.")
-              self ! decorateMessage(
-                StopYarnSession(
-                  FinalApplicationStatus.FAILED,
-                  "Fatal error in AM: The ContainerLaunchContext was not set."))
-              0
-          }
-        case None =>
-          log.error("The NMClient was not set.")
-          self ! decorateMessage(
-            StopYarnSession(
-              FinalApplicationStatus.FAILED,
-              "Fatal error in AM: The NMClient was not set."))
-          0
-      }
-    } else {
-      0
-    }
-  }
 
   private def runningContainerIds(): List[ContainerId] = {
-    runningContainersList map { runningCont => runningCont.getId}
+    runningContainersList map { runningCont => runningCont.getId }
   }
+
   private def allocatedContainerIds(): List[ContainerId] = {
-    allocatedContainersList map { runningCont => runningCont.getId}
+    allocatedContainersList map { runningCont => runningCont.getId }
   }
 
-  /** Starts the Yarn session by connecting to the RessourceManager and the NodeManager.
After
+  /** Starts the Yarn session by connecting to the ResourceManager and the NodeManager. After
     * a connection has been established, the number of missing containers is requested from
Yarn.
     *
     * @param conf Hadoop configuration object
@@ -495,7 +288,7 @@ class YarnJobManager(
           YarnConfiguration.DEFAULT_RM_AM_EXPIRY_INTERVAL_MS),
         MILLISECONDS)
 
-      if(YARN_HEARTBEAT_DELAY.gteq(yarnExpiryInterval)) {
+      if (YARN_HEARTBEAT_DELAY.gteq(yarnExpiryInterval)) {
         log.warn(s"The heartbeat interval of the Flink Application master " +
           s"($YARN_HEARTBEAT_DELAY) is greater than YARN's expiry interval " +
           s"($yarnExpiryInterval). The application is likely to be killed by YARN.")
@@ -516,11 +309,19 @@ class YarnJobManager(
       val shipListString = env.get(FlinkYarnClientBase.ENV_CLIENT_SHIP_FILES)
       val yarnClientUsername = env.get(FlinkYarnClientBase.ENV_CLIENT_USERNAME)
 
-      val rm = AMRMClient.createAMRMClient[ContainerRequest]()
-      rm.init(conf)
-      rm.start()
 
-      rmClientOption = Some(rm)
+      // wrap default client in asynchronous handler thread
+      val rmClientAsync: AMRMClientAsync[ContainerRequest] = AMRMClientAsync.createAMRMClientAsync(
+        FAST_YARN_HEARTBEAT_DELAY.toMillis.toInt,
+        AMRMClientAsyncHandler)
+
+      // inject client into handler to adjust the heartbeat interval and make requests
+      AMRMClientAsyncHandler.setClient(rmClientAsync)
+
+      rmClientAsync.init(conf)
+      rmClientAsync.start()
+
+      rmClientOption = Some(rmClientAsync)
 
       val nm = NMClient.createNMClient()
       nm.init(conf)
@@ -535,7 +336,7 @@ class YarnJobManager(
 
       val actorSystemPort = AkkaUtils.getAddress(system).port.getOrElse(-1)
 
-      val response = rm.registerApplicationMaster(
+      val response = rmClientAsync.registerApplicationMaster(
         applicationMasterHost,
         actorSystemPort,
         url)
@@ -555,7 +356,7 @@ class YarnJobManager(
           log.info(s"Requesting initial TaskManager container $i.")
           numPendingRequests += 1
           // these are initial requests. The reallocation setting doesn't affect this.
-          rm.addContainerRequest(containerRequest)
+          rmClientAsync.addContainerRequest(containerRequest)
       }
 
       val flinkJar = Records.newRecord(classOf[LocalResource])
@@ -601,10 +402,6 @@ class YarnJobManager(
           taskManagerLocalResources)
       )
 
-      context.system.scheduler.scheduleOnce(
-        FAST_YARN_HEARTBEAT_DELAY,
-        self,
-        decorateMessage(HeartbeatWithYarn))
     } recover {
       case t: Throwable =>
         log.error("Could not start yarn session.", t)
@@ -622,24 +419,11 @@ class YarnJobManager(
     *                 living containers.
     * @return Seq of living containers which could be retrieved
     */
-  private def getContainersFromPreviousAttempts(
-      response: RegisterApplicationMasterResponse)
-    : Seq[Container] = {
-
+  private def getContainersFromPreviousAttempts(response: RegisterApplicationMasterResponse)
+      : Seq[Container] = {
     RegisterApplicationMasterResponseReflector.getContainersFromPreviousAttempts(response)
   }
 
-  private def tryToReturnContainers(returnRequest: Set[PreemptionContainer]): Unit = {
-    for(requestedBackContainers <- returnRequest) {
-      allocatedContainersList = allocatedContainersList.dropWhile( container => {
-        val result = requestedBackContainers.getId.equals(container.getId)
-        if(result) {
-          log.info(s"Returning container $container back to ResourceManager.")
-        }
-        result
-      })
-    }
-  }
 
   private def getContainerRequest(memoryPerTaskManager: Int): ContainerRequest = {
     // Priority for worker containers - priorities are intra-application
@@ -659,7 +443,7 @@ class YarnJobManager(
       hasLog4j: Boolean,
       yarnClientUsername: String,
       yarnConf: Configuration,
-      taskManagerLocalResources: Map[String, LocalResource]) : ContainerLaunchContext = {
+      taskManagerLocalResources: Map[String, LocalResource]): ContainerLaunchContext = {
     log.info("Create container launch context.")
     val ctx = Records.newRecord(classOf[ContainerLaunchContext])
 
@@ -736,7 +520,7 @@ class YarnJobManager(
     val useOffHeap = flinkConfiguration.getBoolean(
       ConfigConstants.TASK_MANAGER_MEMORY_OFF_HEAP_KEY, false)
 
-    if (useOffHeap && eagerAllocation){
+    if (useOffHeap && eagerAllocation) {
       val fixedOffHeapSize = flinkConfiguration.getLong(
         ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, -1L)
 
@@ -754,6 +538,248 @@ class YarnJobManager(
       memoryLimit
     }
   }
+
+  /**
+    * Heartbeats with the resource manager and handles container updates.
+    */
+  object AMRMClientAsyncHandler extends AMRMClientAsync.CallbackHandler {
+
+    /*
+     * Asynchronous client to make requests to the RM.
+     * Must be set via setClient(..) before its service is started.
+     */
+    private var client : AMRMClientAsync[ContainerRequest] = null
+
+    override def onError(e: Throwable): Unit = {
+      self ! decorateMessage(
+        StopYarnSession(
+          FinalApplicationStatus.FAILED,
+          "Error in communication with Yarn resource manager: " + e.getMessage)
+      )
+    }
+
+    override def getProgress: Float = {
+      runningContainers.toFloat / numTaskManagers
+    }
+
+    override def onShutdownRequest(): Unit = {
+      // We are getting killed anyway
+    }
+
+    override def onNodesUpdated(updatedNodes: JavaList[NodeReport]): Unit = {
+      // We are not interested in node updates
+    }
+
+    override def onContainersCompleted(statuses: JavaList[ContainerStatus]): Unit = {
+
+      val completedContainerStatuses = statuses.asScala
+      val idStatusMap = completedContainerStatuses
+        .map(status => (status.getContainerId, status)).toMap
+
+      completedContainerStatuses.foreach {
+        status => log.info(s"Container ${status.getContainerId} is completed " +
+          s"with diagnostics: ${status.getDiagnostics}")
+      }
+
+      // get failed containers (returned containers are also completed, so we have to
+      // distinguish if it was running before).
+      val (completedContainers, remainingRunningContainers) = runningContainersList
+        .partition(idStatusMap contains _.getId)
+
+      completedContainers.foreach {
+        container =>
+          val status = idStatusMap(container.getId)
+          failedContainers += 1
+          runningContainers -= 1
+          log.info(s"Container ${status.getContainerId} was a running container. " +
+            s"Total failed containers $failedContainers.")
+          val detail = status.getExitStatus match {
+            case -103 => "Vmem limit exceeded";
+            case -104 => "Pmem limit exceeded";
+            case _ => ""
+          }
+          messageListener foreach {
+            _ ! decorateMessage(
+              YarnMessage(s"Diagnostics for containerID=${status.getContainerId} in " +
+                s"state=${status.getState}.\n${status.getDiagnostics} $detail")
+            )
+          }
+      }
+
+      runningContainersList = remainingRunningContainers
+
+      // maxFailedContainers == -1 is infinite number of retries.
+      if (maxFailedContainers != -1 && failedContainers >= maxFailedContainers)
{
+        val msg = s"Stopping YARN session because the number of failed " +
+          s"containers ($failedContainers) exceeded the maximum failed container " +
+          s"count ($maxFailedContainers). This number is controlled by " +
+          s"the '${ConfigConstants.YARN_MAX_FAILED_CONTAINERS}' configuration " +
+          s"setting. By default its the number of requested containers"
+        log.error(msg)
+        self ! decorateMessage(StopYarnSession(FinalApplicationStatus.FAILED, msg))
+
+      }
+
+      allocateContainers()
+
+    }
+
+    override def onContainersAllocated(containers: JavaList[Container]): Unit = {
+
+      val newlyAllocatedContainers = containers.asScala
+
+      newlyAllocatedContainers.foreach {
+        container => log.info(s"Got new container for allocation: ${container.getId}")
+      }
+
+      allocatedContainersList ++= containers.asScala
+      numPendingRequests = math.max(0, numPendingRequests - newlyAllocatedContainers.length)
+
+      allocateContainers()
+
+      if (runningContainers >= numTaskManagers && allocatedContainersList.nonEmpty)
{
+        log.info(s"Flink has ${allocatedContainersList.size} allocated containers which "
+
+          s"are not needed right now. Returning them")
+        for (container <- allocatedContainersList) {
+          client.releaseAssignedContainer(container.getId)
+        }
+        allocatedContainersList = List()
+      }
+    }
+
+    /**
+      * Allocates new containers if necessary.
+      */
+    private def allocateContainers() : Unit = {
+
+      // check if we want to start some of our allocated containers.
+      if (runningContainers < numTaskManagers) {
+        val missingContainers = numTaskManagers - runningContainers
+        log.info(s"The user requested $numTaskManagers containers, $runningContainers " +
+          s"running. $missingContainers containers missing")
+
+        val numStartedContainers = startTMsInAllocatedContainers(missingContainers)
+
+        // if there are still containers missing, request them from YARN
+        val toAllocateFromYarn = Math.max(
+          missingContainers - numStartedContainers - numPendingRequests,
+          0)
+
+        if (toAllocateFromYarn > 0) {
+          val reallocate = flinkConfiguration
+            .getBoolean(ConfigConstants.YARN_REALLOCATE_FAILED_CONTAINERS, true)
+          log.info(s"There are $missingContainers containers missing." +
+            s" $numPendingRequests are already requested. " +
+            s"Requesting $toAllocateFromYarn additional container(s) from YARN. " +
+            s"Reallocation of failed containers is enabled=$reallocate " +
+            s"('${ConfigConstants.YARN_REALLOCATE_FAILED_CONTAINERS}')")
+          // there are still containers missing. Request them from YARN
+          if (reallocate) {
+            for (i <- 1 to toAllocateFromYarn) {
+              val containerRequest = getContainerRequest(memoryPerTaskManager)
+              client.addContainerRequest(containerRequest)
+              numPendingRequests += 1
+              log.info("Requested additional container from YARN. Pending requests " +
+                s"$numPendingRequests.")
+            }
+          }
+        }
+      }
+    }
+
+    /** Starts min(numTMsToStart, allocatedContainersList.size) TaskManager in the available
+      * allocated containers. The number of successfully started TaskManagers is returned.
+      *
+      * @param numTMsToStart Number of TaskManagers to start if enough allocated containers
are
+      *                      available. If not, then all allocated containers are used
+      * @return Number of successfully started TaskManagers
+      */
+    private def startTMsInAllocatedContainers(numTMsToStart: Int): Int = {
+      // not enough containers running
+      if (allocatedContainersList.nonEmpty) {
+        log.info(s"${allocatedContainersList.size} containers already allocated by YARN.
" +
+          "Starting...")
+
+        nmClientOption match {
+          case Some(nmClient) =>
+            containerLaunchContext match {
+              case Some(ctx) =>
+                val (containersToBeStarted, remainingContainers) = allocatedContainersList
+                  .splitAt(numTMsToStart)
+
+                val startedContainers = containersToBeStarted.flatMap {
+                  container =>
+                    try {
+                      nmClient.startContainer(container, ctx)
+                      val message = s"Launching container (${container.getId} on host " +
+                        s"${container.getNodeId.getHost})."
+                      log.info(message)
+
+                      messageListener foreach {
+                        _ ! decorateMessage(YarnMessage(message))
+                      }
+
+                      Some(container)
+                    } catch {
+                      case e: YarnException =>
+                        log.error(s"Exception while starting YARN " +
+                          s"container ${container.getId} on " +
+                          s"host ${container.getNodeId.getHost}", e)
+                        None
+                    }
+                }
+
+                runningContainers += startedContainers.length
+                runningContainersList :::= startedContainers
+
+                allocatedContainersList = remainingContainers
+
+                if (runningContainers < numTaskManagers) {
+                  setHeartbeatRate(FAST_YARN_HEARTBEAT_DELAY)
+                } else {
+                  setHeartbeatRate(YARN_HEARTBEAT_DELAY)
+                }
+
+                startedContainers.length
+              case None =>
+                log.error("The ContainerLaunchContext was not set.")
+                self ! decorateMessage(
+                  StopYarnSession(
+                    FinalApplicationStatus.FAILED,
+                    "Fatal error in AM: The ContainerLaunchContext was not set."))
+                0
+            }
+          case None =>
+            log.error("The NMClient was not set.")
+            self ! decorateMessage(
+              StopYarnSession(
+                FinalApplicationStatus.FAILED,
+                "Fatal error in AM: The NMClient was not set."))
+            0
+        }
+      } else {
+        0
+      }
+    }
+
+    /**
+      * Adjusts the heartbeat interval of the asynchronous client.
+      * @param interval The interval between the heartbeats.
+      */
+    private def setHeartbeatRate(interval : FiniteDuration): Unit = {
+      client.setHeartbeatInterval(interval.toMillis.toInt)
+    }
+
+    /**
+      * Register the client with the CallbackHandler. Must be called before the client is
started.
+      * @param clientAsync The AMRM client to make requests with.
+      */
+    def setClient(clientAsync: AMRMClientAsync[ContainerRequest]) = {
+      client = clientAsync
+    }
+
+  }
+
 }
 
 /** Singleton object to reflect the getContainersFromPreviousAttempts method for


Mime
View raw message