TEZ-4007: Introduce AmExtensions and Zookeeper-based FrameworkServices - checkstyle, spotbugs, javadoc improvements, refactor, test fixes

Author: abstractdog

pom.xml

@@ -69,7 +69,7 @@
     <commons-io.version>2.16.0</commons-io.version>
     <commons-lang.version>2.6</commons-lang.version>
     <clover.license>${user.home}/clover.license</clover.license>
-    <curator.version>2.7.1</curator.version>
+    <curator.version>5.9.0</curator.version>
     <dependency-check-maven.version>3.2.0</dependency-check-maven.version>
     <dependency-maven-plugin.version>3.8.1</dependency-maven-plugin.version>
     <spotbugs.version>4.9.3</spotbugs.version>
@@ -749,6 +749,18 @@
         <type>test-jar</type>
         <scope>test</scope>
       </dependency>
+      <dependency>
+        <groupId>org.apache.curator</groupId>
+        <artifactId>curator-test</artifactId>
+        <version>${curator.version}</version>
+        <scope>test</scope>
+        <exclusions>
+          <exclusion>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+          </exclusion>
+        </exclusions>
+      </dependency>
       <dependency>
         <groupId>org.mockito</groupId>
         <artifactId>mockito-core</artifactId>

tez-api/findbugs-exclude.xml

@@ -151,4 +151,17 @@
     <Method name="getApplicationId" />
     <Bug pattern="EI_EXPOSE_REP" />
   </Match>
+
+  <!-- TEZ-4007 -->
+  <Match>
+    <Class name="org.apache.tez.client.registry.zookeeper.ZkFrameworkClient" />
+    <Field name="amPort" />
+    <Bug pattern="AT_STALE_THREAD_WRITE_OF_PRIMITIVE" />
+  </Match>
+
+  <Match>
+    <Class name="org.apache.tez.client.TezYarnClient" />
+    <Method name="&lt;init&gt;" params="org.apache.hadoop.yarn.client.api.YarnClient" returns="void" />
+    <Bug pattern="EI_EXPOSE_REP2" />
+  </Match>
 </FindBugsFilter>

tez-api/pom.xml

@@ -126,7 +126,6 @@
     <dependency>
       <groupId>org.apache.curator</groupId>
       <artifactId>curator-test</artifactId>
-      <version>${curator.version}</version>
       <scope>test</scope>
     </dependency>
   </dependencies>

tez-api/src/main/java/org/apache/tez/client/FrameworkClient.java

@@ -19,15 +19,13 @@
 package org.apache.tez.client;
 
 import java.io.IOException;
-import java.util.Optional;
 
 import org.apache.hadoop.classification.InterfaceAudience.Private;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ApplicationReport;
 import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
-import org.apache.hadoop.yarn.client.api.YarnClient;
 import org.apache.hadoop.yarn.client.api.YarnClientApplication;
 import org.apache.hadoop.yarn.exceptions.YarnException;
 import org.apache.tez.common.RPCUtil;
@@ -49,6 +47,7 @@
 import org.apache.tez.dag.api.client.rpc.DAGClientAMProtocolRPC.SubmitDAGResponseProto;
 import org.apache.tez.frameworkplugins.ClientFrameworkService;
 import org.apache.tez.frameworkplugins.FrameworkUtils;
+import org.apache.tez.frameworkplugins.yarn.YarnClientFrameworkService;
 
 import com.google.protobuf.ServiceException;
 
@@ -60,21 +59,18 @@ public abstract class FrameworkClient {
   protected static final Logger LOG = LoggerFactory.getLogger(FrameworkClient.class);
 
   public static FrameworkClient createFrameworkClient(TezConfiguration tezConf) {
-    Optional<FrameworkClient> pluginClient =
-        FrameworkUtils.get(ClientFrameworkService.class, tezConf)
-            .flatMap(framework -> framework.createOrGetFrameworkClient(tezConf));
-
     boolean isLocal = tezConf.getBoolean(TezConfiguration.TEZ_LOCAL_MODE, TezConfiguration.TEZ_LOCAL_MODE_DEFAULT);
     if (isLocal) {
       try {
         return ReflectionUtils.createClazzInstance("org.apache.tez.client.LocalClient");
       } catch (TezReflectionException e) {
         throw new TezUncheckedException("Fail to create LocalClient", e);
       }
-    } else if (pluginClient.isPresent()) {
-      return pluginClient.get();
+    } else {
+      ClientFrameworkService clientFrameworkService = FrameworkUtils.get(ClientFrameworkService.class, tezConf);
+      return clientFrameworkService == null ? new YarnClientFrameworkService().newFrameworkClient()
+          : clientFrameworkService.newFrameworkClient();
     }
-    return new TezYarnClient(YarnClient.createYarnClient());
   }
 
   /**

tez-api/src/main/java/org/apache/tez/client/TezClient.java

@@ -672,8 +672,6 @@ private DAGClient submitDAGSession(DAG dag) throws TezException, IOException {
     }
 
     Map<String, LocalResource> tezJarResources = getTezJarResources(sessionCredentials);
-
-
     DAGPlan dagPlan = TezClientUtils.prepareAndCreateDAGPlan(dag, amConfig, tezJarResources,
         usingTezArchiveDeploy, sessionCredentials, servicePluginsDescriptor, javaOptsChecker);
 
@@ -688,7 +686,6 @@ private DAGClient submitDAGSession(DAG dag) throws TezException, IOException {
 
     // if request size exceeds maxSubmitDAGRequestSizeThroughIPC, we serialize them to HDFS
     SubmitDAGRequestProto request = requestBuilder.build();
-
     if (request.getSerializedSize() > maxSubmitDAGRequestSizeThroughIPC) {
       Path dagPlanPath = new Path(TezCommonUtils.getTezSystemStagingPath(amConfig.getTezConfiguration(),
           sessionAppId.toString()), TezConstants.TEZ_PB_PLAN_BINARY_NAME +

tez-api/src/main/java/org/apache/tez/client/TezClientUtils.java

@@ -149,8 +149,6 @@ private static Path getPath(String configUri) {
     }
   }
 
-
-
   /**
    * Setup LocalResource map for Tez jars based on provided Configuration
    *
@@ -188,7 +186,7 @@ static boolean setupTezJarsLocalResources(TezConfiguration conf,
           + conf.get(TezConfiguration.TEZ_LIB_URIS_CLASSPATH));
 
       usingTezArchive = addLocalResources(conf, tezJarUris,
-          tezJarResources, credentials);
+                         tezJarResources, credentials);
 
       if (tezJarResources.isEmpty()) {
         throw new TezUncheckedException(
@@ -265,8 +263,8 @@ private static boolean addLocalResources(Configuration conf,
 
         // Add URI fragment or just the filename
         Path name = new Path((null == u.getFragment())
-            ? p.getName()
-            : u.getFragment());
+          ? p.getName()
+          : u.getFragment());
         if (name.isAbsolute()) {
           throw new IllegalArgumentException("Resource name must be "
               + "relative, not absolute: " + name
@@ -579,7 +577,7 @@ public static ApplicationSubmissionContext createApplicationSubmissionContext(
     // don't overwrite existing conf, needed for TezClient.getClient() so existing containers have stable resource fingerprints
     if(!binaryConfPath.getFileSystem(tezConf).exists(binaryConfPath)) {
       ConfigurationProto finalConfProto = createFinalConfProtoForApp(tezConf,
-          servicePluginsDescriptor);
+              servicePluginsDescriptor);
 
       FSDataOutputStream amConfPBOutBinaryStream = null;
       try {
@@ -620,12 +618,12 @@ public static ApplicationSubmissionContext createApplicationSubmissionContext(
     }
 
     LocalResource sessionJarsPBLRsrc =
-        TezClientUtils.createLocalResource(fs,
-            sessionJarsPath, LocalResourceType.FILE,
-            LocalResourceVisibility.APPLICATION);
+      TezClientUtils.createLocalResource(fs,
+        sessionJarsPath, LocalResourceType.FILE,
+        LocalResourceVisibility.APPLICATION);
     amLocalResources.put(
-        TezConstants.TEZ_AM_LOCAL_RESOURCES_PB_FILE_NAME,
-        sessionJarsPBLRsrc);
+      TezConstants.TEZ_AM_LOCAL_RESOURCES_PB_FILE_NAME,
+      sessionJarsPBLRsrc);
 
     String user = UserGroupInformation.getCurrentUser().getShortUserName();
     ACLManager aclManager = new ACLManager(user, amConfig.getTezConfiguration());
@@ -657,9 +655,9 @@ public static ApplicationSubmissionContext createApplicationSubmissionContext(
       }
 
       amLocalResources.put(TezConstants.TEZ_PB_PLAN_BINARY_NAME,
-          TezClientUtils.createLocalResource(fs,
-              binaryPath, LocalResourceType.FILE,
-              LocalResourceVisibility.APPLICATION));
+        TezClientUtils.createLocalResource(fs,
+          binaryPath, LocalResourceType.FILE,
+          LocalResourceVisibility.APPLICATION));
 
       if (Level.DEBUG.isGreaterOrEqual(Level.toLevel(amLogLevel))) {
         Path textPath = localizeDagPlanAsText(dagPB, fs, amConfig, strAppId, tezSysStagingPath);
@@ -690,7 +688,7 @@ public static ApplicationSubmissionContext createApplicationSubmissionContext(
 
     Collection<String> tagsFromConf =
         amConfig.getTezConfiguration().getTrimmedStringCollection(
-            TezConfiguration.TEZ_APPLICATION_TAGS);
+        TezConfiguration.TEZ_APPLICATION_TAGS);
 
     appContext.setApplicationType(TezConstants.TEZ_APPLICATION_TYPE);
     if (tagsFromConf != null && !tagsFromConf.isEmpty()) {
@@ -711,8 +709,8 @@ public static ApplicationSubmissionContext createApplicationSubmissionContext(
     appContext.setAMContainerSpec(amContainer);
 
     appContext.setMaxAppAttempts(
-        amConfig.getTezConfiguration().getInt(TezConfiguration.TEZ_AM_MAX_APP_ATTEMPTS,
-            TezConfiguration.TEZ_AM_MAX_APP_ATTEMPTS_DEFAULT));
+      amConfig.getTezConfiguration().getInt(TezConfiguration.TEZ_AM_MAX_APP_ATTEMPTS,
+        TezConfiguration.TEZ_AM_MAX_APP_ATTEMPTS_DEFAULT));
 
     return appContext;
 
@@ -850,7 +848,7 @@ public static void addLog4jSystemProperties(String logLevel,
   }
 
   public static ConfigurationProto createFinalConfProtoForApp(Configuration amConf,
-      ServicePluginsDescriptor servicePluginsDescriptor) {
+    ServicePluginsDescriptor servicePluginsDescriptor) {
     assert amConf != null;
     ConfigurationProto.Builder builder = ConfigurationProto.newBuilder();
     for (Entry<String, String> entry : amConf) {
@@ -951,7 +949,7 @@ static DAGClientAMProtocolBlockingPB getAMProxy(FrameworkClient frameworkClient,
               + ", trackingUrl=" + appReport.getTrackingUrl()
               + ", diagnostics="
               + (appReport.getDiagnostics() != null ? appReport.getDiagnostics()
-              : TezClient.NO_CLUSTER_DIAGNOSTICS_MSG);
+                  : TezClient.NO_CLUSTER_DIAGNOSTICS_MSG);
           LOG.info(msg);
           throw new SessionNotRunning(msg);
         }
@@ -1017,7 +1015,7 @@ static void createSessionToken(String tokenIdentifier,
   public static String maybeAddDefaultMemoryJavaOpts(String javaOpts, Resource resource,
       double maxHeapFactor) {
     if ((javaOpts != null && !javaOpts.isEmpty()
-        && (javaOpts.contains("-Xmx") || javaOpts.contains("-Xms")))
+          && (javaOpts.contains("-Xmx") || javaOpts.contains("-Xms")))
         || (resource.getMemory() <= 0)) {
       return javaOpts;
     }
@@ -1028,8 +1026,8 @@ public static String maybeAddDefaultMemoryJavaOpts(String javaOpts, Resource res
 
     if (Double.parseDouble("-1") == maxHeapFactor) {
       maxHeapFactor = resource.getMemory() < TezConstants.TEZ_CONTAINER_SMALL_SLAB_BOUND_MB
-          ? TezConstants.TEZ_CONTAINER_MAX_JAVA_HEAP_FRACTION_SMALL_SLAB
-          : TezConstants.TEZ_CONTAINER_MAX_JAVA_HEAP_FRACTION_LARGE_SLAB;
+        ? TezConstants.TEZ_CONTAINER_MAX_JAVA_HEAP_FRACTION_SMALL_SLAB
+        : TezConstants.TEZ_CONTAINER_MAX_JAVA_HEAP_FRACTION_LARGE_SLAB;
     }
     int maxMemory = (int)(resource.getMemory() * maxHeapFactor);
     maxMemory = maxMemory <= 0 ? 1 : maxMemory;
@@ -1039,7 +1037,7 @@ public static String maybeAddDefaultMemoryJavaOpts(String javaOpts, Resource res
   }
 
   private static boolean checkAncestorPermissionsForAllUsers(Configuration conf, Path pathComponent,
-      FsAction permission) throws IOException {
+                                                             FsAction permission) throws IOException {
     FileSystem fs = pathComponent.getFileSystem(conf);
 
     if (Shell.WINDOWS && fs instanceof LocalFileSystem) {

tez-api/src/main/java/org/apache/tez/client/TezYarnClient.java

@@ -43,7 +43,7 @@ public class TezYarnClient extends FrameworkClient {
   private String amHost;
   private int amPort;
 
-  protected TezYarnClient(YarnClient yarnClient) {
+  public TezYarnClient(YarnClient yarnClient) {
     this.yarnClient = yarnClient;
   }
 

tez-api/src/main/java/org/apache/tez/client/registry/AMRecord.java

@@ -156,6 +156,11 @@ public ServiceRecord toServiceRecord() {
     return serviceRecord;
   }
 
+  @Override
+  public String toString() {
+    return toServiceRecord().attributes().toString();
+  }
+
   @Override
   public int hashCode() {
     return Objects.hash(appId, host, port, externalId);

tez-api/src/main/java/org/apache/tez/client/registry/AMRegistry.java

@@ -20,7 +20,6 @@
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Optional;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.service.AbstractService;
@@ -31,19 +30,28 @@
 import org.slf4j.LoggerFactory;
 
 /**
- * Base class for AMRegistry implementation
- * Implementation class is configured by tez.am.registry.class
- * Implementations should implement relevant service lifecycle operations:
- *   init, serviceStart, serviceStop, etc..
+ * Base class for {@code AMRegistry} implementations.
  *
- *  init/serviceStart will be invoked during DAGAppMaster.serviceInit
+ * <p>The specific implementation is configured via the
+ * {@code tez.am.registry.class} property.</p>
  *
- *  serviceStop will invoked on DAGAppMaster shutdown
+ * <p>Implementations are expected to provide appropriate service lifecycle
+ * behavior, including:
+ * <ul>
+ *   <li>{@code init}</li>
+ *   <li>{@code serviceStart}</li>
+ *   <li>{@code serviceStop}</li>
+ * </ul>
+ * </p>
+ *
+ * <p>{@code init} and {@code serviceStart} are invoked during
+ * {@code DAGAppMaster.serviceInit()}, while {@code serviceStop} is called
+ * when {@code DAGAppMaster} shuts down.</p>
  */
 public abstract class AMRegistry extends AbstractService {
 
   private static final Logger LOG = LoggerFactory.getLogger(AMRegistry.class);
-  protected List<AMRecord> amRecords = new ArrayList<>();
+  private List<AMRecord> amRecords = new ArrayList<>();
 
   @Override
   public void init(Configuration conf) {
@@ -79,13 +87,14 @@ public void add(AMRecord server) throws Exception {
 
   public abstract void remove(AMRecord server) throws Exception;
 
-  public Optional<ApplicationId> generateNewId() throws Exception {
-    return Optional.empty();
+  public ApplicationId generateNewId() throws Exception {
+    return null;
   }
 
   public abstract AMRecord createAmRecord(ApplicationId appId, String hostName, int port);
 
-  @Override public void serviceStop() throws Exception {
+  @Override
+  public void serviceStop() throws Exception {
     List<AMRecord> records = new ArrayList<>(amRecords);
     for(AMRecord record : records) {
       remove(record);