[5/7] flink git commit: [FLINK-3545][yarn] integrate ResourceManager support

[mx...@apache.org]

http://git-wip-us.apache.org/repos/asf/flink/blob/4405235e/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnMessages.scala
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnMessages.scala b/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnMessages.scala
index 7a35590..8645581 100644
--- a/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnMessages.scala
+++ b/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnMessages.scala
@@ -29,17 +29,10 @@ import scala.concurrent.duration.{Deadline, FiniteDuration}
 
 object YarnMessages {
 
-  case class YarnMessage(message: String, date: Date = new Date())
   case class ApplicationMasterStatus(numTaskManagers: Int, numSlots: Int)
 
-  case object UnregisterClient extends RequiresLeaderSessionID
-
-  case class StopYarnSession(status: FinalApplicationStatus, diagnostics: String)
-    extends RequiresLeaderSessionID
   case class LocalStopYarnSession(status: FinalApplicationStatus, diagnostics: String)
 
-  case object JobManagerStopped
-
   /**
     * Entry point to start a new YarnSession.
     * @param config The configuration to start the YarnSession with.
@@ -48,13 +41,13 @@ object YarnMessages {
   case class StartYarnSession(config: Configuration, webServerPort: Int)
 
   /**
-    * Callback from the async resource manager client when containers were allocated.
+    * Callback from the async ResourceManager client when containers were allocated.
     * @param containers List of containers which were allocated.
     */
   case class YarnContainersAllocated(containers: JavaList[Container])
 
   /**
-    * Callback from the async resource manager client when containers were completed.
+    * Callback from the async ResourceManager client when containers were completed.
     * @param statuses List of the completed containers' status.
     */
   case class YarnContainersCompleted(statuses: JavaList[ContainerStatus])
@@ -70,8 +63,8 @@ object YarnMessages {
       currentTimeout: FiniteDuration,
       deadline: Option[Deadline]) extends RequiresLeaderSessionID
 
-  /** Registration message sent from the [[ApplicationClient]] to the [[YarnJobManager]]. A
-    * succesful registration is acknowledged with a [[AcknowledgeApplicationClientRegistration]]
+  /** Registration message sent from the [[ApplicationClient]] to the [[YarnFlinkResourceManager]].
+    * A successful registration is acknowledged with a [[AcknowledgeApplicationClientRegistration]]
     * message.
     */
   case object RegisterApplicationClient extends RequiresLeaderSessionID
@@ -90,12 +83,9 @@ object YarnMessages {
   case class JobManagerLeaderAddress(jobManagerAkkaURL: String, leaderSessionID: UUID)
 
   case object HeartbeatWithYarn
-  case object PollYarnClusterStatus // see org.apache.flink.runtime.yarn.FlinkYarnClusterStatus for
-                                    // the response
   case object CheckForUserCommand
 
   // tell the AM to monitor the job and stop once it has finished
-  case class StopAMAfterJob(jobId:JobID) extends RequiresLeaderSessionID
   case class LocalStopAMAfterJob(jobId:JobID)
 
   case object LocalGetYarnMessage // request new message

http://git-wip-us.apache.org/repos/asf/flink/blob/4405235e/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnProcessShutDownThread.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnProcessShutDownThread.java b/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnProcessShutDownThread.java
deleted file mode 100644
index 8e36ac0..0000000
--- a/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnProcessShutDownThread.java
+++ /dev/null
@@ -1,82 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.yarn;
-
-import akka.actor.ActorSystem;
-import org.slf4j.Logger;
-import scala.concurrent.duration.Duration;
-
-import java.util.concurrent.TimeoutException;
-
-import static com.google.common.base.Preconditions.checkNotNull;
-
-/**
- * JVM shut down thread awaiting actor system shut down for a certain amount
- * of time before exiting the JVM.
- *
- * <p>On some Linux distributions, YARN is not able to stop containers, because
- * the <code>kill</code> command has different arguments. For example when
- * running Flink on GCE ("Debian GNU/Linux 7.9 (wheezy)"), YARN containers will
- * not properly shut down when we don't call <code>System.exit()</code>.
- */
-class YarnProcessShutDownThread extends Thread {
-
-	/** Log of the corresponding YARN process. */
-	private final Logger log;
-
-	/** Actor system to await termination of. */
-	private final ActorSystem actorSystem;
-
-	/** Actor system termination timeout before shutting down the JVM. */
-	private final Duration terminationTimeout;
-
-	/**
-	 * Creates a shut down thread.
-	 *
-	 * @param log                Log of the corresponding YARN process.
-	 * @param actorSystem        Actor system to await termination of.
-	 * @param terminationTimeout Actor system termination timeout before
-	 *                           shutting down the JVM.
-	 */
-	public YarnProcessShutDownThread(
-			Logger log,
-			ActorSystem actorSystem,
-			Duration terminationTimeout) {
-
-		this.log = checkNotNull(log, "Logger");
-		this.actorSystem = checkNotNull(actorSystem, "Actor system");
-		this.terminationTimeout = checkNotNull(terminationTimeout, "Termination timeout");
-	}
-
-	@Override
-	public void run() {
-		try {
-			actorSystem.awaitTermination(terminationTimeout);
-		} catch (Exception e) {
-			if (e instanceof TimeoutException) {
-				log.error("Actor system shut down timed out.", e);
-			} else {
-				log.error("Failure during actor system shut down.", e);
-			}
-		} finally {
-			log.info("Shutdown completed. Stopping JVM.");
-			System.exit(0);
-		}
-	}
-}

http://git-wip-us.apache.org/repos/asf/flink/blob/4405235e/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnTaskManager.scala
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnTaskManager.scala b/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnTaskManager.scala
index 828441b..8b39f52 100644
--- a/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnTaskManager.scala
+++ b/flink-yarn/src/main/scala/org/apache/flink/yarn/YarnTaskManager.scala
@@ -18,13 +18,15 @@
 
 package org.apache.flink.yarn
 
+import org.apache.flink.runtime.clusterframework.messages.StopCluster
+import org.apache.flink.runtime.clusterframework.types.ResourceID
 import org.apache.flink.runtime.instance.InstanceConnectionInfo
 import org.apache.flink.runtime.io.disk.iomanager.IOManager
 import org.apache.flink.runtime.io.network.NetworkEnvironment
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService
 import org.apache.flink.runtime.memory.MemoryManager
 import org.apache.flink.runtime.taskmanager.{TaskManagerConfiguration, TaskManager}
-import org.apache.flink.yarn.YarnMessages.StopYarnSession
+import org.apache.flink.runtime.util.ProcessShutDownThread
 
 import scala.concurrent.duration._
 
@@ -33,6 +35,7 @@ import scala.concurrent.duration._
   */
 class YarnTaskManager(
     config: TaskManagerConfiguration,
+    resourceID: ResourceID,
     connectionInfo: InstanceConnectionInfo,
     memoryManager: MemoryManager,
     ioManager: IOManager,
@@ -41,6 +44,7 @@ class YarnTaskManager(
     leaderRetrievalService: LeaderRetrievalService)
   extends TaskManager(
     config,
+    resourceID,
     connectionInfo,
     memoryManager,
     ioManager,
@@ -49,19 +53,17 @@ class YarnTaskManager(
     leaderRetrievalService) {
 
   override def handleMessage: Receive = {
-    handleYarnMessages orElse super.handleMessage
+    super.handleMessage
   }
+}
 
-  def handleYarnMessages: Receive = {
-    case StopYarnSession(status, diagnostics) =>
-      log.info(s"Stopping YARN TaskManager with final application status $status " +
-        s"and diagnostics: $diagnostics")
-      context.system.shutdown()
+  object YarnTaskManager {
+    /** Entry point (main method) to run the TaskManager on YARN.
+      *
+      * @param args The command line arguments.
+      */
+    def main(args: Array[String]): Unit = {
+      YarnTaskManagerRunner.runYarnTaskManager(args, classOf[YarnTaskManager])
+    }
 
-      // Await actor system termination and shut down JVM
-      new YarnProcessShutDownThread(
-        log.logger,
-        context.system,
-        FiniteDuration(10, SECONDS)).start()
-  }
 }