YARN-6572. Refactoring Router services to use common util classes for pipeline creations.

Author: slfan1989

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/utils/YarnServerBuilderUtils.java

@@ -79,7 +79,7 @@ public static NodeHeartbeatResponse newNodeHeartbeatResponse(int responseId,
    *
    * @param applicationId Application ID
    * @param credentials HDFS Tokens
-   * @return systemCredentialsForAppsProto SystemCredentialsForAppsProto
+   * @return systemCredentialsForAppsProto
    */
   public static SystemCredentialsForAppsProto newSystemCredentialsForAppsProto(
       ApplicationId applicationId, ByteBuffer credentials) {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/RouterServerUtil.java

@@ -18,13 +18,29 @@
 
 package org.apache.hadoop.yarn.server.router;
 
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.commons.lang3.reflect.MethodUtils;
 import org.apache.hadoop.classification.InterfaceAudience.Private;
 import org.apache.hadoop.classification.InterfaceAudience.Public;
 import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
+import org.apache.hadoop.yarn.server.router.clientrm.ClientMethod;
+import org.apache.hadoop.yarn.server.router.clientrm.ClientRequestInterceptor;
+import org.apache.hadoop.yarn.server.router.rmadmin.RMAdminRequestInterceptor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
 /**
  * Common utility methods used by the Router server.
  *
@@ -60,4 +76,53 @@ public static void logAndThrowException(String errMsg, Throwable t)
     }
   }
 
+  public static <R> R createRequestInterceptorChain(Configuration conf, String pipeLineClassName,
+      String interceptorClassName, ClientMethod request, Class<R> clazz)
+      throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
+
+    List<String> interceptorClassNames = getInterceptorClassNames(conf,
+        pipeLineClassName, interceptorClassName);
+
+    R pipeline = null;
+    R current = null;
+
+    for (String className : interceptorClassNames) {
+      try {
+        Class<?> interceptorClass = conf.getClassByName(className);
+        if (clazz.isAssignableFrom(interceptorClass)) {
+          R interceptorInstance = (R) ReflectionUtils.newInstance(interceptorClass, conf);
+          if (pipeline == null) {
+            pipeline = interceptorInstance;
+            current = interceptorInstance;
+            continue;
+          } else {
+            Method method = clazz.getMethod(request.getMethodName(), request.getTypes());
+            method.invoke(current, interceptorInstance);
+            current = interceptorInstance;
+          }
+        } else {
+          LOG.error("Class: {} not instance of {}.", className, clazz.getCanonicalName());
+          throw new YarnRuntimeException("Class: " + className + " not instance of "
+              + clazz.getCanonicalName());
+        }
+      } catch (ClassNotFoundException e) {
+        throw new YarnRuntimeException("Could not instantiate RequestInterceptor: " + className, e);
+      }
+    }
+
+    return pipeline;
+  }
+
+  private static List<String> getInterceptorClassNames(Configuration conf,
+      String pipeLineClass, String interceptorClass) {
+    String configuredInterceptorClassNames = conf.get(pipeLineClass, interceptorClass);
+    List<String> interceptorClassNames = new ArrayList<String>();
+    Collection<String> tempList =
+        StringUtils.getStringCollection(configuredInterceptorClassNames);
+    for (String item : tempList) {
+      interceptorClassNames.add(item.trim());
+    }
+    return interceptorClassNames;
+  }
+
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/clientrm/RouterClientRMService.java

@@ -19,6 +19,7 @@
 package org.apache.hadoop.yarn.server.router.clientrm;
 
 import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -110,6 +111,7 @@
 import org.apache.hadoop.yarn.exceptions.YarnException;
 import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
 import org.apache.hadoop.yarn.ipc.YarnRPC;
+import org.apache.hadoop.yarn.server.router.RouterServerUtil;
 import org.apache.hadoop.yarn.server.router.security.authorize.RouterPolicyProvider;
 import org.apache.hadoop.yarn.util.LRUCacheHashMap;
 import org.slf4j.Logger;
@@ -147,7 +149,7 @@ public RouterClientRMService() {
 
   @Override
   protected void serviceStart() throws Exception {
-    LOG.info("Starting Router ClientRMService");
+    LOG.info("Starting Router ClientRMService.");
     Configuration conf = getConfig();
     YarnRPC rpc = YarnRPC.create(conf);
     UserGroupInformation.setConfiguration(conf);
@@ -161,9 +163,7 @@ protected void serviceStart() throws Exception {
     int maxCacheSize =
         conf.getInt(YarnConfiguration.ROUTER_PIPELINE_CACHE_MAX_SIZE,
             YarnConfiguration.DEFAULT_ROUTER_PIPELINE_CACHE_MAX_SIZE);
-    this.userPipelineMap = Collections.synchronizedMap(
-        new LRUCacheHashMap<String, RequestInterceptorChainWrapper>(
-            maxCacheSize, true));
+    this.userPipelineMap = Collections.synchronizedMap(new LRUCacheHashMap<>(maxCacheSize, true));
 
     Configuration serverConf = new Configuration(conf);
 
@@ -181,14 +181,13 @@ protected void serviceStart() throws Exception {
     }
 
     this.server.start();
-    LOG.info("Router ClientRMService listening on address: "
-        + this.server.getListenerAddress());
+    LOG.info("Router ClientRMService listening on address: {}.", this.server.getListenerAddress());
     super.serviceStart();
   }
 
   @Override
   protected void serviceStop() throws Exception {
-    LOG.info("Stopping Router ClientRMService");
+    LOG.info("Stopping Router ClientRMService.");
     if (this.server != null) {
       this.server.stop();
     }
@@ -201,27 +200,6 @@ public Server getServer() {
     return this.server;
   }
 
-  /**
-   * Returns the comma separated intercepter class names from the configuration.
-   *
-   * @param conf
-   * @return the intercepter class names as an instance of ArrayList
-   */
-  private List<String> getInterceptorClassNames(Configuration conf) {
-    String configuredInterceptorClassNames =
-        conf.get(YarnConfiguration.ROUTER_CLIENTRM_INTERCEPTOR_CLASS_PIPELINE,
-            YarnConfiguration.DEFAULT_ROUTER_CLIENTRM_INTERCEPTOR_CLASS);
-
-    List<String> interceptorClassNames = new ArrayList<String>();
-    Collection<String> tempList =
-        StringUtils.getStringCollection(configuredInterceptorClassNames);
-    for (String item : tempList) {
-      interceptorClassNames.add(item.trim());
-    }
-
-    return interceptorClassNames;
-  }
-
   @Override
   public GetNewApplicationResponse getNewApplication(
       GetNewApplicationRequest request) throws YarnException, IOException {
@@ -507,42 +485,25 @@ protected Map<String, RequestInterceptorChainWrapper> getPipelines() {
   @VisibleForTesting
   protected ClientRequestInterceptor createRequestInterceptorChain() {
     Configuration conf = getConfig();
+    ClientRequestInterceptor pipeline = null;
+    ClientMethod remoteMethod = null;
+    try {
+      remoteMethod = new ClientMethod("setNextInterceptor",
+           new Class[]{ClientRequestInterceptor.class}, new Object[]{null});
 
-    List<String> interceptorClassNames = getInterceptorClassNames(conf);
+      pipeline = RouterServerUtil.createRequestInterceptorChain(conf,
+              YarnConfiguration.ROUTER_CLIENTRM_INTERCEPTOR_CLASS_PIPELINE,
+              YarnConfiguration.DEFAULT_ROUTER_CLIENTRM_INTERCEPTOR_CLASS,
+              remoteMethod, ClientRequestInterceptor.class);
 
-    ClientRequestInterceptor pipeline = null;
-    ClientRequestInterceptor current = null;
-    for (String interceptorClassName : interceptorClassNames) {
-      try {
-        Class<?> interceptorClass = conf.getClassByName(interceptorClassName);
-        if (ClientRequestInterceptor.class.isAssignableFrom(interceptorClass)) {
-          ClientRequestInterceptor interceptorInstance =
-              (ClientRequestInterceptor) ReflectionUtils
-                  .newInstance(interceptorClass, conf);
-          if (pipeline == null) {
-            pipeline = interceptorInstance;
-            current = interceptorInstance;
-            continue;
-          } else {
-            current.setNextInterceptor(interceptorInstance);
-            current = interceptorInstance;
-          }
-        } else {
-          throw new YarnRuntimeException(
-              "Class: " + interceptorClassName + " not instance of "
-                  + ClientRequestInterceptor.class.getCanonicalName());
-        }
-      } catch (ClassNotFoundException e) {
+      if (pipeline == null) {
         throw new YarnRuntimeException(
-            "Could not instantiate ApplicationClientRequestInterceptor: "
-                + interceptorClassName,
-            e);
+            "RequestInterceptor pipeline is not configured in the system.");
       }
-    }
-
-    if (pipeline == null) {
-      throw new YarnRuntimeException(
-          "RequestInterceptor pipeline is not configured in the system");
+    } catch (IOException | InvocationTargetException | NoSuchMethodException | RuntimeException
+         | IllegalAccessException ex) {
+      throw new YarnRuntimeException("RequestInterceptor pipeline is not configured in the system.",
+          ex);
     }
     return pipeline;
   }
@@ -566,14 +527,14 @@ private RequestInterceptorChainWrapper initializePipeline(String user) {
         // We should init the pipeline instance after it is created and then
         // add to the map, to ensure thread safe.
         LOG.info("Initializing request processing pipeline for application "
-            + "for the user: {}", user);
+            + "for the user: {}.", user);
 
         ClientRequestInterceptor interceptorChain =
             this.createRequestInterceptorChain();
         interceptorChain.init(user);
         chainWrapper.init(interceptorChain);
       } catch (Exception e) {
-        LOG.error("Init ClientRequestInterceptor error for user: " + user, e);
+        LOG.error("Init ClientRequestInterceptor error for user: {}.", user, e);
         throw e;
       }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/rmadmin/RouterRMAdminService.java

@@ -19,6 +19,7 @@
 package org.apache.hadoop.yarn.server.router.rmadmin;
 
 import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -69,6 +70,9 @@
 import org.apache.hadoop.yarn.server.api.protocolrecords.ReplaceLabelsOnNodeResponse;
 import org.apache.hadoop.yarn.server.api.protocolrecords.UpdateNodeResourceRequest;
 import org.apache.hadoop.yarn.server.api.protocolrecords.UpdateNodeResourceResponse;
+import org.apache.hadoop.yarn.server.router.RouterServerUtil;
+import org.apache.hadoop.yarn.server.router.clientrm.ClientMethod;
+import org.apache.hadoop.yarn.server.router.clientrm.ClientRequestInterceptor;
 import org.apache.hadoop.yarn.server.router.security.authorize.RouterPolicyProvider;
 import org.apache.hadoop.yarn.util.LRUCacheHashMap;
 import org.slf4j.Logger;
@@ -164,27 +168,6 @@ public Server getServer() {
     return this.server;
   }
 
-  /**
-   * Returns the comma separated intercepter class names from the configuration.
-   *
-   * @param conf
-   * @return the intercepter class names as an instance of ArrayList
-   */
-  private List<String> getInterceptorClassNames(Configuration conf) {
-    String configuredInterceptorClassNames =
-        conf.get(YarnConfiguration.ROUTER_RMADMIN_INTERCEPTOR_CLASS_PIPELINE,
-            YarnConfiguration.DEFAULT_ROUTER_RMADMIN_INTERCEPTOR_CLASS);
-
-    List<String> interceptorClassNames = new ArrayList<String>();
-    Collection<String> tempList =
-        StringUtils.getStringCollection(configuredInterceptorClassNames);
-    for (String item : tempList) {
-      interceptorClassNames.add(item.trim());
-    }
-
-    return interceptorClassNames;
-  }
-
   @VisibleForTesting
   protected RequestInterceptorChainWrapper getInterceptorChain()
       throws IOException {
@@ -215,43 +198,24 @@ protected Map<String, RequestInterceptorChainWrapper> getPipelines() {
   @VisibleForTesting
   protected RMAdminRequestInterceptor createRequestInterceptorChain() {
     Configuration conf = getConfig();
+    RMAdminRequestInterceptor pipeline = null;
+    ClientMethod remoteMethod = null;
+    try {
+      remoteMethod = new ClientMethod("setNextInterceptor",
+              new Class[]{RMAdminRequestInterceptor.class}, new Object[]{null});
 
-    List<String> interceptorClassNames = getInterceptorClassNames(conf);
+      pipeline = RouterServerUtil.createRequestInterceptorChain(conf,
+          YarnConfiguration.ROUTER_RMADMIN_INTERCEPTOR_CLASS_PIPELINE,
+          YarnConfiguration.DEFAULT_ROUTER_RMADMIN_INTERCEPTOR_CLASS,
+          remoteMethod, RMAdminRequestInterceptor.class);
 
-    RMAdminRequestInterceptor pipeline = null;
-    RMAdminRequestInterceptor current = null;
-    for (String interceptorClassName : interceptorClassNames) {
-      try {
-        Class<?> interceptorClass = conf.getClassByName(interceptorClassName);
-        if (RMAdminRequestInterceptor.class
-            .isAssignableFrom(interceptorClass)) {
-          RMAdminRequestInterceptor interceptorInstance =
-              (RMAdminRequestInterceptor) ReflectionUtils
-                  .newInstance(interceptorClass, conf);
-          if (pipeline == null) {
-            pipeline = interceptorInstance;
-            current = interceptorInstance;
-            continue;
-          } else {
-            current.setNextInterceptor(interceptorInstance);
-            current = interceptorInstance;
-          }
-        } else {
-          throw new YarnRuntimeException(
-              "Class: " + interceptorClassName + " not instance of "
-                  + RMAdminRequestInterceptor.class.getCanonicalName());
-        }
-      } catch (ClassNotFoundException e) {
+      if (pipeline == null) {
         throw new YarnRuntimeException(
-            "Could not instantiate RMAdminRequestInterceptor: "
-                + interceptorClassName,
-            e);
+            "RequestInterceptor pipeline is not configured in the system.");
       }
-    }
-
-    if (pipeline == null) {
-      throw new YarnRuntimeException(
-          "RequestInterceptor pipeline is not configured in the system");
+    } catch (IOException | InvocationTargetException | NoSuchMethodException | RuntimeException
+             | IllegalAccessException ex) {
+      throw new YarnRuntimeException("Create RequestInterceptor Chain error.", ex);
     }
     return pipeline;
   }
@@ -274,14 +238,14 @@ private RequestInterceptorChainWrapper initializePipeline(String user) {
       try {
         // We should init the pipeline instance after it is created and then
         // add to the map, to ensure thread safe.
-        LOG.info("Initializing request processing pipeline for user: {}", user);
+        LOG.info("Initializing request processing pipeline for user: {}.", user);
 
         RMAdminRequestInterceptor interceptorChain =
             this.createRequestInterceptorChain();
         interceptorChain.init(user);
         chainWrapper.init(interceptorChain);
       } catch (Exception e) {
-        LOG.error("Init RMAdminRequestInterceptor error for user: " + user, e);
+        LOG.error("Init RMAdminRequestInterceptor error for user: {}.", user, e);
         throw e;
       }