TEZ-4007: Zookeeper based FrameworkServices and AmExtensions (3/3) - checkstyle, spotbugs, javadoc improvements, refactor

Author: abstractdog

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;
 
@@ -37,7 +37,6 @@
 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,8 +45,8 @@
 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);
@@ -57,7 +56,6 @@ public class ZkAMRegistryClient extends AMRegistryClient {
   //Cache of known AMs
   private ConcurrentHashMap<String, AMRecord> amRecordCache = new ConcurrentHashMap<>();
   private CuratorFramework client;
-  private PathChildrenCache cache;
 
   private static Map<String, ZkAMRegistryClient> INSTANCES = new HashMap<>();
 
@@ -79,7 +77,7 @@ private ZkAMRegistryClient(final Configuration conf) {
   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,7 +89,15 @@ public void start() throws Exception {
     cache.getListenable().addListener(new ZkRegistryListener());
   }
 
-  //Deserialize ServiceRecord from Zookeeper to populate AMRecord in cache
+  /**
+   * 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
@@ -104,75 +110,76 @@ public static AMRecord getAMRecord(final ChildData childData) throws IOException
     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 synchronized void addListener(AMRegistryClientListener listener) {
-    listeners.add(listener);
+  @Override
+  public List<AMRecord> getAllRecords() {
+    return amRecordCache.values().stream().map(AMRecord::new).collect(Collectors.toList());
   }
 
-  //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());
-        }
       }
     }
 

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

@@ -37,15 +37,13 @@
 public class ZkFrameworkClient extends FrameworkClient {
 
   private AMRecord amRecord;
-  private TezConfiguration tezConf;
   private ZkAMRegistryClient amRegistryClient = null;
   private boolean isRunning = false;
   private String amHost;
   private int amPort;
 
   @Override
   public synchronized void init(TezConfiguration tezConf) {
-    this.tezConf = tezConf;
     if (this.amRegistryClient == null) {
       try {
         this.amRegistryClient = ZkAMRegistryClient.getClient(tezConf);

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

@@ -0,0 +1,24 @@
+/**
+ * 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.
+ */
+
+@Public
+@Evolving
+package org.apache.tez.client.registry.zookeeper;
+
+import org.apache.hadoop.classification.InterfaceAudience.Public;
+import org.apache.hadoop.classification.InterfaceStability.Evolving;

tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java

@@ -869,8 +869,7 @@ public TezConfiguration(boolean loadDefaults) {
   /** Int value. Port used for AM RPC*/
   @ConfigurationScope(Scope.AM)
   @ConfigurationProperty(type="integer")
-  public static final String TEZ_AM_RPC_PORT =
-    TEZ_AM_PREFIX + "rpc.port";
+  public static final String TEZ_AM_RPC_PORT = TEZ_AM_PREFIX + "rpc.port";
   public static final int TEZ_AM_RPC_PORT_DEFAULT = 0;
 
   /** Int value. Number of threads to handle client RPC requests. Expert level setting.*/
@@ -2351,52 +2350,57 @@ static Set<String> getPropertySet() {
   public static final String TEZ_TASK_ATTEMPT_HOOKS = TEZ_TASK_PREFIX + "attempt.hooks";
 
   /**
-   * String value
+   * String value. ZooKeeper quorum connection string used when creating a CuratorFramework for the ZooKeeper registry.
    */
   @ConfigurationScope(Scope.AM)
   @ConfigurationProperty
   public static final String TEZ_AM_ZOOKEEPER_QUORUM = TEZ_AM_PREFIX + "zookeeper.quorum";
 
   /**
-   * String value
+   * String value. Namespace in ZooKeeper registry for the Application Master.
    */
   @ConfigurationScope(Scope.AM)
   @ConfigurationProperty
   public static final String TEZ_AM_REGISTRY_NAMESPACE = TEZ_AM_PREFIX + "registry.namespace";
   public static final String TEZ_AM_REGISTRY_NAMESPACE_DEFAULT = "/tez_am/server";
 
   /**
-   * Integer value
+   * Integer value. Initial backoff sleep duration (in milliseconds) for Curator retries.
+   * It's used when creating a CuratorFramework for the ZooKeeper registry.
    */
   @ConfigurationScope(Scope.AM)
   @ConfigurationProperty
   public static final String TEZ_AM_CURATOR_BACKOFF_SLEEP = TEZ_AM_PREFIX + "curator.backoff.sleep";
   public static final int TEZ_AM_CURATOR_BACKOFF_SLEEP_DEFAULT = 1000;
 
   /**
-   * Integer value
+   * Integer value. Maximum number of retries for Curator operations.
+   * It's used when creating a CuratorFramework for the ZooKeeper registry.
    */
   @ConfigurationScope(Scope.AM)
   @ConfigurationProperty
   public static final String TEZ_AM_CURATOR_MAX_RETRIES = TEZ_AM_PREFIX + "curator.max.retries";
   public static final int TEZ_AM_CURATOR_MAX_RETRIES_DEFAULT = 3;
 
   /**
-   * Integer value (milliseconds)
+   * Integer value. Session timeout (in milliseconds) for Curator framework.
+   * It's used when creating a CuratorFramework for the ZooKeeper registry.
    */
   @ConfigurationScope(Scope.AM)
   @ConfigurationProperty
   public static final String TEZ_AM_CURATOR_SESSION_TIMEOUT = TEZ_AM_PREFIX + "curator.session.timeout";
   public static final int TEZ_AM_CURATOR_SESSION_TIMEOUT_DEFAULT = 60000;
 
   /**
-    * Integer value (milliseconds)
-  */
+   * Integer value. Connection timeout (in milliseconds) for Curator framework.
+   * It's used when creating a CuratorFramework for the ZooKeeper registry.
+   */
   @ConfigurationScope(Scope.AM)
   @ConfigurationProperty
   public static final String TEZ_AM_CURATOR_CONNECTION_TIMEOUT = TEZ_AM_PREFIX + "curator.connection.timeout";
   public static final int TEZ_AM_CURATOR_CONNECTION_TIMEOUT_DEFAULT = 15000;
 
+
   @ConfigurationProperty
   public static final String TEZ_FRAMEWORK_MODE = TEZ_PREFIX + ".framework.mode";
 

tez-api/src/main/java/org/apache/tez/frameworkplugins/ClientFrameworkService.java

@@ -21,7 +21,6 @@
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.tez.client.FrameworkClient;
-import org.apache.tez.client.registry.AMRegistryClient;
 
 /*
   FrameworkService that runs code within the client process that is using TezClient
@@ -30,6 +29,4 @@
 public interface ClientFrameworkService extends FrameworkService {
   //Provide an impl. for org.apache.tez.client.FrameworkClient
   Optional<FrameworkClient> createOrGetFrameworkClient(Configuration conf);
-  //Provide an impl. for org.apache.tez.registry.AMRegistryClient
-  Optional<AMRegistryClient> createOrGetRegistryClient(Configuration conf);
 }

tez-api/src/main/java/org/apache/tez/frameworkplugins/FrameworkMode.java

@@ -6,9 +6,9 @@
  * 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
- *
+ * <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.
@@ -28,13 +28,25 @@ public enum FrameworkMode {
 
   STANDALONE_ZOOKEEPER(
       "org.apache.tez.frameworkplugins.zookeeper.ZookeeperStandaloneClientFrameworkService",
-      "org.apache.tez.frameworkplugins.zookeeper.ZookeeperStandaloneServerFrameworkService");
+      "org.apache.tez.frameworkplugins.zookeeper.ZookeeperStandaloneServerFrameworkService"),
+  
+  YARN(
+      "org.apache.tez.frameworkplugins.yarn.YarnClientFrameworkService",
+      "org.apache.tez.frameworkplugins.yarn.YarnServerFrameworkService");
 
-  String clientClassName;
-  String serverClassName;
+  private final String clientClassName;
+  private final String serverClassName;
 
   FrameworkMode(String clientClassName, String serverClassName) {
     this.clientClassName = clientClassName;
     this.serverClassName = serverClassName;
   }
+
+  public String getClientClassName() {
+    return clientClassName;
+  }
+
+  public String getServerClassName() {
+    return serverClassName;
+  }
 }