YARN-824. Added static factory methods to hadoop-yarn-client interfaces. Contributed by Jian He.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1493631 13f79535-47bb-0310-9956-ffa450edef68

Author: acmurthy

hadoop-yarn-project/CHANGES.txt

@@ -169,6 +169,9 @@ Release 2.1.0-beta - UNRELEASED
     YARN-831. Removed minimum resource from GetNewApplicationResponse as a
     follow-up to YARN-787. (Jian He via acmurthy)
 
+    YARN-824. Added static factory methods to hadoop-yarn-client interfaces. 
+    (Jian He via acmurthy)
+
   NEW FEATURES
 
     YARN-482. FS: Extend SchedulingMode to intermediate queues. 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/Client.java

@@ -60,7 +60,7 @@
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.api.records.YarnApplicationState;
 import org.apache.hadoop.yarn.api.records.YarnClusterMetrics;
-import org.apache.hadoop.yarn.client.YarnClientImpl;
+import org.apache.hadoop.yarn.client.YarnClient;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.YarnException;
 import org.apache.hadoop.yarn.util.ConverterUtils;
@@ -99,13 +99,13 @@
  */
 @InterfaceAudience.Public
 @InterfaceStability.Unstable
-public class Client extends YarnClientImpl {
+public class Client {
 
   private static final Log LOG = LogFactory.getLog(Client.class);
 
   // Configuration
   private Configuration conf;
-
+  private YarnClient yarnClient;
   // Application master specific info to register a new Application with RM/ASM
   private String appName = "";
   // App master priority
@@ -186,9 +186,10 @@ public static void main(String[] args) {
   /**
    */
   public Client(Configuration conf) throws Exception  {
-    super();
+    
     this.conf = conf;
-    init(conf);
+    yarnClient = YarnClient.createYarnClient();
+    yarnClient.init(conf);
     opts = new Options();
     opts.addOption("appname", true, "Application Name. Default value - DistributedShell");
     opts.addOption("priority", true, "Application Priority. Default 0");
@@ -317,13 +318,13 @@ public boolean init(String[] args) throws ParseException {
   public boolean run() throws IOException, YarnException {
 
     LOG.info("Running Client");
-    start();
+    yarnClient.start();
 
-    YarnClusterMetrics clusterMetrics = super.getYarnClusterMetrics();
+    YarnClusterMetrics clusterMetrics = yarnClient.getYarnClusterMetrics();
     LOG.info("Got Cluster metric info from ASM" 
         + ", numNodeManagers=" + clusterMetrics.getNumNodeManagers());
 
-    List<NodeReport> clusterNodeReports = super.getNodeReports();
+    List<NodeReport> clusterNodeReports = yarnClient.getNodeReports();
     LOG.info("Got Cluster node info from ASM");
     for (NodeReport node : clusterNodeReports) {
       LOG.info("Got node report from ASM for"
@@ -333,15 +334,15 @@ public boolean run() throws IOException, YarnException {
           + ", nodeNumContainers" + node.getNumContainers());
     }
 
-    QueueInfo queueInfo = super.getQueueInfo(this.amQueue);		
+    QueueInfo queueInfo = yarnClient.getQueueInfo(this.amQueue);
     LOG.info("Queue info"
         + ", queueName=" + queueInfo.getQueueName()
         + ", queueCurrentCapacity=" + queueInfo.getCurrentCapacity()
         + ", queueMaxCapacity=" + queueInfo.getMaximumCapacity()
         + ", queueApplicationCount=" + queueInfo.getApplications().size()
         + ", queueChildQueueCount=" + queueInfo.getChildQueues().size());		
 
-    List<QueueUserACLInfo> listAclInfo = super.getQueueAclsInfo();				
+    List<QueueUserACLInfo> listAclInfo = yarnClient.getQueueAclsInfo();
     for (QueueUserACLInfo aclInfo : listAclInfo) {
       for (QueueACL userAcl : aclInfo.getUserAcls()) {
         LOG.info("User ACL Info for Queue"
@@ -351,7 +352,7 @@ public boolean run() throws IOException, YarnException {
     }		
 
     // Get a new application id 
-    GetNewApplicationResponse newApp = super.getNewApplication();
+    GetNewApplicationResponse newApp = yarnClient.getNewApplication();
     ApplicationId appId = newApp.getApplicationId();
 
     // TODO get min/max resource capabilities from RM and change memory ask if needed
@@ -564,7 +565,7 @@ public boolean run() throws IOException, YarnException {
     // or an exception thrown to denote some form of a failure
     LOG.info("Submitting application to ASM");
 
-    super.submitApplication(appContext);
+    yarnClient.submitApplication(appContext);
 
     // TODO
     // Try submitting the same request again
@@ -596,7 +597,7 @@ private boolean monitorApplication(ApplicationId appId)
       }
 
       // Get application report for the appId we are interested in 
-      ApplicationReport report = super.getApplicationReport(appId);
+      ApplicationReport report = yarnClient.getApplicationReport(appId);
 
       LOG.info("Got application report from ASM for"
           + ", appId=" + appId.getId()
@@ -656,7 +657,7 @@ private void forceKillApplication(ApplicationId appId)
 
     // Response can be ignored as it is non-null on success or 
     // throws an exception in case of failures
-    super.killApplication(appId);	
+    yarnClient.killApplication(appId);	
   }
 
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-unmanaged-am-launcher/src/main/java/org/apache/hadoop/yarn/applications/unmanagedamlauncher/UnmanagedAMLauncher.java

@@ -48,6 +48,7 @@
 import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
 import org.apache.hadoop.yarn.api.records.Priority;
 import org.apache.hadoop.yarn.api.records.YarnApplicationState;
+import org.apache.hadoop.yarn.client.YarnClient;
 import org.apache.hadoop.yarn.client.YarnClientImpl;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.YarnException;
@@ -72,7 +73,7 @@ public class UnmanagedAMLauncher {
   private Configuration conf;
 
   // Handle to talk to the Resource Manager/Applications Manager
-  private YarnClientImpl rmClient;
+  private YarnClient rmClient;
 
   // Application master specific info to register a new Application with RM/ASM
   private String appName = "";
@@ -160,7 +161,7 @@ public boolean init(String[] args) throws ParseException {
     }
 
     YarnConfiguration yarnConf = new YarnConfiguration(conf);
-    rmClient = new YarnClientImpl();
+    rmClient = YarnClient.createYarnClient();
     rmClient.init(yarnConf);
 
     return true;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/AMRMClient.java

@@ -24,22 +24,48 @@
 import java.util.concurrent.ConcurrentMap;
 
 import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.classification.InterfaceAudience.Public;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
 import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
 import org.apache.hadoop.yarn.api.records.Priority;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.api.records.Token;
 import org.apache.hadoop.yarn.exceptions.YarnException;
-import org.apache.hadoop.yarn.service.Service;
+import org.apache.hadoop.yarn.service.AbstractService;
 
 import com.google.common.collect.ImmutableList;
 
 @InterfaceAudience.Public
 @InterfaceStability.Unstable
-public interface AMRMClient<T extends AMRMClient.ContainerRequest> extends Service {
+public abstract class AMRMClient<T extends AMRMClient.ContainerRequest> extends
+    AbstractService {
+
+  /**
+   * Create a new instance of AMRMClient.
+   * For usage:
+   * <pre>
+   * {@code
+   * AMRMClient.<T>createAMRMClientContainerRequest(appAttemptId)
+   * }</pre>
+   * @param appAttemptId the appAttemptId associated with the AMRMClient
+   * @return the newly create AMRMClient instance.
+   */
+  @Public
+  public static <T extends ContainerRequest> AMRMClient<T> createAMRMClient(
+      ApplicationAttemptId appAttemptId) {
+    AMRMClient<T> client = new AMRMClientImpl<T>(appAttemptId);
+    return client;
+  }
+
+  @Private
+  protected AMRMClient(String name) {
+    super(name);
+  }
 
   /**
    * Object to represent container request for resources. Scheduler
@@ -132,7 +158,7 @@ public StoredContainerRequest(Resource capability, String[] nodes,
    * @throws YarnException
    * @throws IOException
    */
-  public RegisterApplicationMasterResponse 
+  public abstract RegisterApplicationMasterResponse 
                registerApplicationMaster(String appHostName,
                                          int appHostPort,
                                          String appTrackingUrl) 
@@ -153,7 +179,7 @@ public StoredContainerRequest(Resource capability, String[] nodes,
    * @throws YarnException
    * @throws IOException
    */
-  public AllocateResponse allocate(float progressIndicator) 
+  public abstract AllocateResponse allocate(float progressIndicator) 
                            throws YarnException, IOException;
 
   /**
@@ -164,7 +190,7 @@ public AllocateResponse allocate(float progressIndicator)
    * @throws YarnException
    * @throws IOException
    */
-  public void unregisterApplicationMaster(FinalApplicationStatus appStatus,
+  public abstract void unregisterApplicationMaster(FinalApplicationStatus appStatus,
                                            String appMessage,
                                            String appTrackingUrl) 
                throws YarnException, IOException;
@@ -173,7 +199,7 @@ public void unregisterApplicationMaster(FinalApplicationStatus appStatus,
    * Request containers for resources before calling <code>allocate</code>
    * @param req Resource request
    */
-  public void addContainerRequest(T req);
+  public abstract void addContainerRequest(T req);
 
   /**
    * Remove previous container request. The previous container request may have 
@@ -182,7 +208,7 @@ public void unregisterApplicationMaster(FinalApplicationStatus appStatus,
    * even after the remove request
    * @param req Resource request
    */
-  public void removeContainerRequest(T req);
+  public abstract void removeContainerRequest(T req);
 
   /**
    * Release containers assigned by the Resource Manager. If the app cannot use
@@ -191,21 +217,21 @@ public void unregisterApplicationMaster(FinalApplicationStatus appStatus,
    * it still needs it. eg. it released non-local resources
    * @param containerId
    */
-  public void releaseAssignedContainer(ContainerId containerId);
+  public abstract void releaseAssignedContainer(ContainerId containerId);
 
   /**
    * Get the currently available resources in the cluster.
    * A valid value is available after a call to allocate has been made
    * @return Currently available resources
    */
-  public Resource getClusterAvailableResources();
+  public abstract Resource getClusterAvailableResources();
 
   /**
    * Get the current number of nodes in the cluster.
    * A valid values is available after a call to allocate has been made
    * @return Current number of nodes in the cluster
    */
-  public int getClusterNodeCount();
+  public abstract int getClusterNodeCount();
 
   /**
    * Get outstanding <code>StoredContainerRequest</code>s matching the given 
@@ -218,7 +244,7 @@ public void unregisterApplicationMaster(FinalApplicationStatus appStatus,
    * collection, requests will be returned in the same order as they were added.
    * @return Collection of request matching the parameters
    */
-  public List<? extends Collection<T>> getMatchingRequests(
+  public abstract List<? extends Collection<T>> getMatchingRequests(
                                            Priority priority, 
                                            String resourceName, 
                                            Resource capability);
@@ -231,5 +257,5 @@ public List<? extends Collection<T>> getMatchingRequests(
    * communicating with NodeManager (ex. NMClient) using NMTokens. If a new
    * NMToken is received for the same node manager then it will be replaced. 
    */
-  public ConcurrentMap<String, Token> getNMTokens();
+  public abstract ConcurrentMap<String, Token> getNMTokens();
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/AMRMClientAsync.java

@@ -120,7 +120,7 @@ public AMRMClientAsync(ApplicationAttemptId id, int intervalMs,
 
   @Private
   @VisibleForTesting
-  public AMRMClientAsync(AMRMClient<T> client, int intervalMs,
+  protected AMRMClientAsync(AMRMClient<T> client, int intervalMs,
       CallbackHandler callbackHandler) {
     super(AMRMClientAsync.class.getName());
     this.client = client;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/AMRMClientImpl.java

@@ -64,18 +64,15 @@
 import org.apache.hadoop.yarn.factories.RecordFactory;
 import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
 import org.apache.hadoop.yarn.ipc.YarnRPC;
-import org.apache.hadoop.yarn.service.AbstractService;
 import org.apache.hadoop.yarn.util.RackResolver;
 
-import com.google.common.base.Joiner;
-
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Joiner;
 
 // TODO check inputs for null etc. YARN-654
 
 @Unstable
-public class AMRMClientImpl<T extends ContainerRequest> 
-                          extends AbstractService implements AMRMClient<T> {
+public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
 
   private static final Log LOG = LogFactory.getLog(AMRMClientImpl.class);
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/NMClient.java

@@ -23,6 +23,8 @@
 import java.util.Map;
 
 import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.classification.InterfaceAudience.Public;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.yarn.api.records.Container;
 import org.apache.hadoop.yarn.api.records.ContainerId;
@@ -31,11 +33,34 @@
 import org.apache.hadoop.yarn.api.records.NodeId;
 import org.apache.hadoop.yarn.api.records.Token;
 import org.apache.hadoop.yarn.exceptions.YarnException;
-import org.apache.hadoop.yarn.service.Service;
+import org.apache.hadoop.yarn.service.AbstractService;
 
 @InterfaceAudience.Public
 @InterfaceStability.Unstable
-public interface NMClient extends Service {
+public abstract class NMClient extends AbstractService {
+
+  /**
+   * Create a new instance of NMClient.
+   */
+  @Public
+  public static NMClient createNMClient() {
+    NMClient client = new NMClientImpl();
+    return client;
+  }
+
+  /**
+   * Create a new instance of NMClient.
+   */
+  @Public
+  public static NMClient createNMClient(String name) {
+    NMClient client = new NMClientImpl(name);
+    return client;
+  }
+
+  @Private
+  protected NMClient(String name) {
+    super(name);
+  }
 
   /**
    * <p>Start an allocated container.</p>
@@ -54,7 +79,7 @@ public interface NMClient extends Service {
    * @throws YarnException
    * @throws IOException
    */
-  Map<String, ByteBuffer> startContainer(Container container,
+  public abstract Map<String, ByteBuffer> startContainer(Container container,
       ContainerLaunchContext containerLaunchContext)
           throws YarnException, IOException;
 
@@ -68,7 +93,7 @@ Map<String, ByteBuffer> startContainer(Container container,
    * @throws YarnException
    * @throws IOException
    */
-  void stopContainer(ContainerId containerId, NodeId nodeId,
+  public abstract void stopContainer(ContainerId containerId, NodeId nodeId,
       Token containerToken) throws YarnException, IOException;
 
   /**
@@ -82,7 +107,7 @@ void stopContainer(ContainerId containerId, NodeId nodeId,
    * @throws YarnException
    * @throws IOException
    */
-  ContainerStatus getContainerStatus(ContainerId containerId, NodeId nodeId,
+  public abstract ContainerStatus getContainerStatus(ContainerId containerId, NodeId nodeId,
       Token containerToken) throws YarnException, IOException;
 
   /**
@@ -92,6 +117,6 @@ ContainerStatus getContainerStatus(ContainerId containerId, NodeId nodeId,
    *
    * @param enabled whether the feature is enabled or not
    */
-  void cleanupRunningContainersOnStop(boolean enabled);
+  public abstract void cleanupRunningContainersOnStop(boolean enabled);
 
 }