TEZ-4007: Zookeeper based FrameworkServices and AmExtensions (3/3) - 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>

tez-api/findbugs-exclude.xml

@@ -151,4 +151,11 @@
     <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>
 </FindBugsFilter>

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;
@@ -60,21 +58,16 @@ 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 {
+      return FrameworkUtils.get(ClientFrameworkService.class, tezConf).newFrameworkClient();
     }
-    return new TezYarnClient(YarnClient.createYarnClient());
   }
 
   /**

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/AMRegistryClient.java

@@ -24,7 +24,7 @@
 import java.util.List;
 
 /**
- * Interface for client-side AM discovery
+ * Interface for client-side AM discovery.
  */
 public abstract class AMRegistryClient implements Closeable {
 

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

@@ -1,20 +1,20 @@
 /**
-  * 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.
-*/
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.tez.client.registry.zookeeper;
 
@@ -31,13 +31,13 @@
 import org.apache.curator.framework.recipes.cache.PathChildrenCache;
 import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
 import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
+import org.apache.curator.shaded.com.google.common.base.Charsets;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.registry.client.binding.RegistryUtils;
 import org.apache.hadoop.registry.client.types.ServiceRecord;
 import org.apache.tez.client.registry.AMRecord;
 import org.apache.tez.client.registry.AMRegistryClient;
-import org.apache.tez.client.registry.AMRegistryClientListener;
 import org.apache.tez.dag.api.TezConfiguration;
 
 import com.google.common.base.Preconditions;
@@ -46,20 +46,20 @@
 import org.slf4j.LoggerFactory;
 
 /**
-  * Curator/Zookeeper impl of AMRegistryClient
-*/
+ * Curator/Zookeeper implementation of {@link AMRegistryClient}.
+ */
 @InterfaceAudience.Public
 public class ZkAMRegistryClient extends AMRegistryClient {
   private static final Logger LOG = LoggerFactory.getLogger(ZkAMRegistryClient.class);
-
+  private static final Map<String, ZkAMRegistryClient> INSTANCES = new HashMap<>();
   private final Configuration conf;
-
   //Cache of known AMs
-  private ConcurrentHashMap<String, AMRecord> amRecordCache = new ConcurrentHashMap<>();
+  private final ConcurrentHashMap<String, AMRecord> amRecordCache = new ConcurrentHashMap<>();
   private CuratorFramework client;
-  private PathChildrenCache cache;
 
-  private static Map<String, ZkAMRegistryClient> INSTANCES = new HashMap<>();
+  private ZkAMRegistryClient(final Configuration conf) {
+    this.conf = conf;
+  }
 
   public static synchronized ZkAMRegistryClient getClient(final Configuration conf) {
     String namespace = conf.get(TezConfiguration.TEZ_AM_REGISTRY_NAMESPACE);
@@ -72,14 +72,31 @@ public static synchronized ZkAMRegistryClient getClient(final Configuration conf
     return registry;
   }
 
-  private ZkAMRegistryClient(final Configuration conf) {
-    this.conf = conf;
+  /**
+   * Deserializes a {@link ServiceRecord} from ZooKeeper data and converts it into an {@link AMRecord}
+   * for caching.
+   *
+   * @param childData the ZooKeeper node data containing a serialized {@link ServiceRecord}
+   * @return an {@link AMRecord} constructed from the deserialized {@link ServiceRecord}, or {@code null}
+   * if no data is present
+   * @throws IOException if the data cannot be deserialized into a {@link ServiceRecord}
+   */
+  public static AMRecord getAMRecord(final ChildData childData) throws IOException {
+    byte[] data = childData.getData();
+    // only the path appeared, there is no data yet
+    if (data.length == 0) {
+      return null;
+    }
+    String value = new String(data, Charsets.UTF_8);
+    RegistryUtils.ServiceRecordMarshal marshal = new RegistryUtils.ServiceRecordMarshal();
+    ServiceRecord serviceRecord = marshal.fromJson(value);
+    return new AMRecord(serviceRecord);
   }
 
   public void start() throws Exception {
     ZkConfig zkConf = new ZkConfig(this.conf);
     client = zkConf.createCuratorFramework();
-    cache = new PathChildrenCache(client, zkConf.getZkNamespace(), true);
+    PathChildrenCache cache = new PathChildrenCache(client, zkConf.getZkNamespace(), true);
     client.start();
     cache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
     for (ChildData childData : cache.getCurrentData()) {
@@ -91,98 +108,85 @@ public void start() throws Exception {
     cache.getListenable().addListener(new ZkRegistryListener());
   }
 
-  //Deserialize ServiceRecord from Zookeeper to populate AMRecord in cache
-  public static AMRecord getAMRecord(final ChildData childData) throws IOException {
-    byte[] data = childData.getData();
-    // only the path appeared, there is no data yet
-    if (data.length == 0) {
-      return null;
-    }
-    String value = new String(data);
-    RegistryUtils.ServiceRecordMarshal marshal = new RegistryUtils.ServiceRecordMarshal();
-    ServiceRecord serviceRecord = marshal.fromJson(value);
-    return new AMRecord(serviceRecord);
-  }
-
-  @Override public AMRecord getRecord(String appId) {
+  @Override
+  public AMRecord getRecord(String appId) {
     if (amRecordCache.get(appId) == null) {
       return null;
     }
     //Return a copy
     return new AMRecord(amRecordCache.get(appId));
   }
 
-  @Override public List<AMRecord> getAllRecords() {
-    return amRecordCache.values().stream()
-        .map(record -> new AMRecord(record)).collect(Collectors.toList());
-    }
+  @Override
+  public List<AMRecord> getAllRecords() {
+    return amRecordCache.values().stream().map(AMRecord::new).collect(Collectors.toList());
+  }
 
-  @Override public synchronized void addListener(AMRegistryClientListener listener) {
-    listeners.add(listener);
+  @Override
+  public void close() {
+    client.close();
   }
 
-  //Callback for Zookeeper to update local cache
+  /**
+   * Callback listener for ZooKeeper events that updates the local cache
+   * when child nodes under the monitored path change.
+   */
   private class ZkRegistryListener implements PathChildrenCacheListener {
 
-  @Override public void childEvent(final CuratorFramework client, final PathChildrenCacheEvent event)
-      throws Exception {
-    Preconditions.checkArgument(client != null && client.getState() == CuratorFrameworkState.STARTED,
+    @Override
+    public void childEvent(final CuratorFramework clientParam, final PathChildrenCacheEvent event) throws Exception {
+      Preconditions.checkArgument(clientParam != null && clientParam.getState() == CuratorFrameworkState.STARTED,
           "Curator client is not started");
 
-    ChildData childData = event.getData();
-    switch (event.getType()) {
-      case CHILD_ADDED:
-        if(isEmpty(childData)) {
-          LOG.info("AppId allocated: {}", childData.getPath());
-        } else {
-          AMRecord amRecord = getAMRecord(childData);
-          if (amRecord != null) {
-            LOG.info("AM registered with data: {}. Notifying {} listeners.", amRecord, listeners.size());
-            amRecordCache.put(amRecord.getApplicationId().toString(), amRecord);
-            notifyOnAdded(amRecord);
+      ChildData childData = event.getData();
+      switch (event.getType()) {
+        case CHILD_ADDED:
+          if (isEmpty(childData)) {
+            LOG.info("AppId allocated: {}", childData.getPath());
+          } else {
+            AMRecord amRecord = getAMRecord(childData);
+            if (amRecord != null) {
+              LOG.info("AM registered with data: {}. Notifying {} listeners.", amRecord, listeners.size());
+              amRecordCache.put(amRecord.getApplicationId().toString(), amRecord);
+              notifyOnAdded(amRecord);
+            }
+          }
+          break;
+        case CHILD_UPDATED:
+          if (isEmpty(childData)) {
+            throw new RuntimeException("AM updated with empty data");
+          } else {
+            AMRecord amRecord = getAMRecord(childData);
+            if (amRecord != null) {
+              LOG.info("AM updated data: {}. Notifying {} listeners.", amRecord, listeners.size());
+              amRecordCache.put(amRecord.getApplicationId().toString(), amRecord);
+              notifyOnAdded(amRecord);
+            }
           }
-        }
-        break;
-      case CHILD_UPDATED:
-        if(isEmpty(childData)) {
-          throw new RuntimeException("AM updated with empty data");
-        } else {
-          AMRecord amRecord = getAMRecord(childData);
-          if (amRecord != null) {
-            LOG.info("AM updated data: {}. Notifying {} listeners.", amRecord, listeners.size());
-            amRecordCache.put(amRecord.getApplicationId().toString(), amRecord);
-            notifyOnAdded(amRecord);
+          break;
+        case CHILD_REMOVED:
+          if (isEmpty(childData)) {
+            LOG.info("Unused AppId unregistered: {}", childData.getPath());
+          } else {
+            AMRecord amRecord = getAMRecord(childData);
+            if (amRecord != null) {
+              LOG.info("AM removed: {}. Notifying {} listeners.", amRecord, listeners.size());
+              amRecordCache.remove(amRecord.getApplicationId().toString(), amRecord);
+              notifyOnRemoved(amRecord);
+            }
           }
-        }
-        break;
-      case CHILD_REMOVED:
-        if(isEmpty(childData)) {
-          LOG.info("Unused AppId unregistered: {}", childData.getPath());
-        } else {
-          AMRecord amRecord = getAMRecord(childData);
-          if (amRecord != null) {
-            LOG.info("AM removed: {}. Notifying {} listeners.", amRecord, listeners.size());
-            amRecordCache.remove(amRecord.getApplicationId().toString(), amRecord);
-            notifyOnRemoved(amRecord);
+          break;
+        default:
+          if (childData == null) {
+            LOG.info("Ignored event {}", event.getType());
+          } else {
+            LOG.info("Ignored event {} for {}", event.getType(), childData.getPath());
           }
-        }
-        break;
-      default:
-        if(childData == null) {
-          LOG.info("Ignored event {}", event.getType());
-        } else {
-          LOG.info("Ignored event {} for {}", event.getType(), childData.getPath());
-        }
       }
     }
 
     private boolean isEmpty(ChildData childData) {
       return childData == null || childData.getData() == null || childData.getData().length == 0;
     }
   }
-
-  @Override
-  public void close() {
-    client.close();
-  }
 }