Merge branch 'trunk' into YARN-11433-V2

Author: slfan1989

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/metrics/RpcMetrics.java

@@ -141,7 +141,7 @@ public static RpcMetrics create(Server server, Configuration conf) {
   MutableCounterLong rpcAuthorizationSuccesses;
   @Metric("Number of client backoff requests")
   MutableCounterLong rpcClientBackoff;
-  @Metric("Number of Slow RPC calls")
+  @Metric("Number of slow RPC calls")
   MutableCounterLong rpcSlowCalls;
   @Metric("Number of requeue calls")
   MutableCounterLong rpcRequeueCalls;

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/curator/ZKCuratorManager.java

@@ -549,7 +549,17 @@ public HadoopZookeeperFactory(String zkPrincipal, String kerberosPrincipal,
     public ZooKeeper newZooKeeper(String connectString, int sessionTimeout,
         Watcher watcher, boolean canBeReadOnly
     ) throws Exception {
-      ZKClientConfig zkClientConfig = new ZKClientConfig();
+      return this.newZooKeeper(connectString, sessionTimeout,
+      watcher, canBeReadOnly, new ZKClientConfig());
+    }
+
+    @Override
+    public ZooKeeper newZooKeeper(String connectString, int sessionTimeout,
+        Watcher watcher, boolean canBeReadOnly, ZKClientConfig zkClientConfig
+    ) throws Exception {
+      if (zkClientConfig == null) {
+        zkClientConfig = new ZKClientConfig();
+      }
       if (zkPrincipal != null) {
         LOG.info("Configuring zookeeper to use {} as the server principal",
             zkPrincipal);

hadoop-common-project/hadoop-common/src/site/markdown/Metrics.md

@@ -74,16 +74,21 @@ The default timeunit used for RPC metrics is milliseconds (as per the below desc
 | `SentBytes` | Total number of sent bytes |
 | `RpcQueueTimeNumOps` | Total number of RPC calls |
 | `RpcQueueTimeAvgTime` | Average queue time in milliseconds |
-| `RpcLockWaitTimeNumOps` | Total number of RPC call (same as RpcQueueTimeNumOps) |
+| `RpcLockWaitTimeNumOps` | Total number of RPC calls (same as RpcQueueTimeNumOps) |
 | `RpcLockWaitTimeAvgTime` | Average time waiting for lock acquisition in milliseconds |
 | `RpcProcessingTimeNumOps` | Total number of RPC calls (same to RpcQueueTimeNumOps) |
 | `RpcProcessingAvgTime` | Average Processing time in milliseconds |
+| `DeferredRpcProcessingTimeNumOps` | Total number of Deferred RPC calls |
+| `DeferredRpcProcessingAvgTime` | Average Deferred Processing time in milliseconds |
+| `RpcResponseTimeNumOps` | Total number of RPC calls (same to RpcQueueTimeNumOps) |
+| `RpcResponseAvgTime` | Average Response time in milliseconds |
 | `RpcAuthenticationFailures` | Total number of authentication failures |
 | `RpcAuthenticationSuccesses` | Total number of authentication successes |
 | `RpcAuthorizationFailures` | Total number of authorization failures |
 | `RpcAuthorizationSuccesses` | Total number of authorization successes |
 | `RpcClientBackoff` | Total number of client backoff requests |
 | `RpcSlowCalls` | Total number of slow RPC calls |
+| `RpcRequeueCalls` | Total number of requeue RPC calls |
 | `RpcCallsSuccesses` | Total number of RPC calls that are successfully processed |
 | `NumOpenConnections` | Current number of open connections |
 | `NumInProcessHandler` | Current number of handlers on working |
@@ -107,6 +112,18 @@ The default timeunit used for RPC metrics is milliseconds (as per the below desc
 | `rpcLockWaitTime`*num*`s90thPercentileLatency` | Shows the 90th percentile of RPC lock wait time in milliseconds (*num* seconds granularity) if `rpc.metrics.quantile.enable` is set to true. *num* is specified by `rpc.metrics.percentiles.intervals`. |
 | `rpcLockWaitTime`*num*`s95thPercentileLatency` | Shows the 95th percentile of RPC lock wait time in milliseconds (*num* seconds granularity) if `rpc.metrics.quantile.enable` is set to true. *num* is specified by `rpc.metrics.percentiles.intervals`. |
 | `rpcLockWaitTime`*num*`s99thPercentileLatency` | Shows the 99th percentile of RPC lock wait time in milliseconds (*num* seconds granularity) if `rpc.metrics.quantile.enable` is set to true. *num* is specified by `rpc.metrics.percentiles.intervals`. |
+| `rpcResponseTime`*num*`sNumOps` | Shows total number of RPC calls (*num* seconds granularity) if `rpc.metrics.quantile.enable` is set to true. *num* is specified by `rpc.metrics.percentiles.intervals`. |
+| `rpcResponseTime`*num*`s50thPercentileLatency` | Shows the 50th percentile of RPC response time in milliseconds (*num* seconds granularity) if `rpc.metrics.quantile.enable` is set to true. *num* is specified by `rpc.metrics.percentiles.intervals`. |
+| `rpcResponseTime`*num*`s75thPercentileLatency` | Shows the 75th percentile of RPC response time in milliseconds (*num* seconds granularity) if `rpc.metrics.quantile.enable` is set to true. *num* is specified by `rpc.metrics.percentiles.intervals`. |
+| `rpcResponseTime`*num*`s90thPercentileLatency` | Shows the 90th percentile of RPC response time in milliseconds (*num* seconds granularity) if `rpc.metrics.quantile.enable` is set to true. *num* is specified by `rpc.metrics.percentiles.intervals`. |
+| `rpcResponseTime`*num*`s95thPercentileLatency` | Shows the 95th percentile of RPC response time in milliseconds (*num* seconds granularity) if `rpc.metrics.quantile.enable` is set to true. *num* is specified by `rpc.metrics.percentiles.intervals`. |
+| `rpcResponseTime`*num*`s99thPercentileLatency` | Shows the 99th percentile of RPC response time in milliseconds (*num* seconds granularity) if `rpc.metrics.quantile.enable` is set to true. *num* is specified by `rpc.metrics.percentiles.intervals`. |
+| `deferredRpcProcessingTime`*num*`sNumOps` | Shows total number of Deferred RPC calls (*num* seconds granularity) if `rpc.metrics.quantile.enable` is set to true. *num* is specified by `rpc.metrics.percentiles.intervals`. |
+| `deferredRpcProcessingTime`*num*`s50thPercentileLatency` | Shows the 50th percentile of Deferred RPC processing time in milliseconds (*num* seconds granularity) if `rpc.metrics.quantile.enable` is set to true. *num* is specified by `rpc.metrics.percentiles.intervals`. |
+| `deferredRpcProcessingTime`*num*`s75thPercentileLatency` | Shows the 75th percentile of Deferred RPC processing time in milliseconds (*num* seconds granularity) if `rpc.metrics.quantile.enable` is set to true. *num* is specified by `rpc.metrics.percentiles.intervals`. |
+| `deferredRpcProcessingTime`*num*`s90thPercentileLatency` | Shows the 90th percentile of Deferred RPC processing time in milliseconds (*num* seconds granularity) if `rpc.metrics.quantile.enable` is set to true. *num* is specified by `rpc.metrics.percentiles.intervals`. |
+| `deferredRpcProcessingTime`*num*`s95thPercentileLatency` | Shows the 95th percentile of Deferred RPC processing time in milliseconds (*num* seconds granularity) if `rpc.metrics.quantile.enable` is set to true. *num* is specified by `rpc.metrics.percentiles.intervals`. |
+| `deferredRpcProcessingTime`*num*`s99thPercentileLatency` | Shows the 99th percentile of Deferred RPC processing time in milliseconds (*num* seconds granularity) if `rpc.metrics.quantile.enable` is set to true. *num* is specified by `rpc.metrics.percentiles.intervals`. |
 | `TotalRequests` | Total num of requests served by the RPC server. |
 | `TotalRequestsPerSeconds` | Total num of requests per second served by the RPC server. |
 

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/curator/TestZKCuratorManager.java

@@ -22,10 +22,15 @@
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
 import javax.security.auth.login.AppConfigurationEntry;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
 import org.apache.curator.test.TestingServer;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.CommonConfigurationKeys;
@@ -193,6 +198,36 @@ public void testJaasConfiguration() throws Exception {
     }
   }
 
+  @Test
+  public void testCuratorFrameworkFactory() throws Exception{
+    // By not explicitly calling the NewZooKeeper method validate that the Curator override works.
+    ZKClientConfig zkClientConfig = new ZKClientConfig();
+    Configuration conf = new Configuration();
+    conf.set(CommonConfigurationKeys.ZK_ADDRESS, this.server.getConnectString());
+    int numRetries = conf.getInt(CommonConfigurationKeys.ZK_NUM_RETRIES,
+        CommonConfigurationKeys.ZK_NUM_RETRIES_DEFAULT);
+    int zkSessionTimeout = conf.getInt(CommonConfigurationKeys.ZK_TIMEOUT_MS,
+        CommonConfigurationKeys.ZK_TIMEOUT_MS_DEFAULT);
+    int zkRetryInterval = conf.getInt(
+        CommonConfigurationKeys.ZK_RETRY_INTERVAL_MS,
+        CommonConfigurationKeys.ZK_RETRY_INTERVAL_MS_DEFAULT);
+    RetryNTimes retryPolicy = new RetryNTimes(numRetries, zkRetryInterval);
+
+    CuratorFramework client = CuratorFrameworkFactory.builder()
+        .connectString(conf.get(CommonConfigurationKeys.ZK_ADDRESS))
+        .zkClientConfig(zkClientConfig)
+        .sessionTimeoutMs(zkSessionTimeout).retryPolicy(retryPolicy)
+        .authorization(new ArrayList<>())
+        .zookeeperFactory(new ZKCuratorManager.HadoopZookeeperFactory(
+            "foo1", "bar1", "bar1.keytab", false,
+            new ZKCuratorManager.TruststoreKeystore(conf))
+
+        ).build();
+    client.start();
+    validateJaasConfiguration(ZKCuratorManager.HadoopZookeeperFactory.JAAS_CLIENT_ENTRY,
+        "bar1", "bar1.keytab", client.getZookeeperClient().getZooKeeper());
+  }
+
   private void validateJaasConfiguration(String clientConfig, String principal, String keytab,
       ZooKeeper zk) {
     assertEquals("Validate that expected clientConfig is set in ZK config", clientConfig,

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSOpsCountStatistics.java

@@ -75,6 +75,7 @@ public enum OpType {
     GET_STORAGE_POLICIES("op_get_storage_policies"),
     GET_STORAGE_POLICY("op_get_storage_policy"),
     GET_TRASH_ROOT("op_get_trash_root"),
+    GET_TRASH_ROOTS("op_get_trash_roots"),
     GET_XATTR("op_get_xattr"),
     LIST_CACHE_DIRECTIVE("op_list_cache_directive"),
     LIST_CACHE_POOL("op_list_cache_pool"),

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java

@@ -3612,6 +3612,8 @@ public Path getTrashRoot(Path path) {
    */
   @Override
   public Collection<FileStatus> getTrashRoots(boolean allUsers) {
+    statistics.incrementReadOps(1);
+    storageStatistics.incrementOpCounter(OpType.GET_TRASH_ROOTS);
     Set<FileStatus> ret = new HashSet<>();
     // Get normal trash roots
     ret.addAll(super.getTrashRoots(allUsers));

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/web/JsonUtilClient.java

@@ -22,6 +22,7 @@
 
 import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.fs.BlockLocation;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.util.Preconditions;
 import org.apache.hadoop.thirdparty.com.google.common.collect.Maps;
 import org.apache.hadoop.fs.ContentSummary;
@@ -877,6 +878,40 @@ public static Map<String, String> getErasureCodeCodecs(Map<?, ?> json) {
     return map;
   }
 
+  public static Collection<FileStatus> getTrashRoots(Map<?, ?> json) {
+    List<?> objs = (List<?>) json.get("Paths");
+    if (objs != null) {
+      FileStatus[] trashRoots = new FileStatus[objs.size()];
+      for (int i = 0; i < objs.size(); i++) {
+        Map<?, ?> m = (Map<?, ?>) objs.get(i);
+        trashRoots[i] = toFileStatus(m);
+      }
+      return Arrays.asList(trashRoots);
+    }
+    return new ArrayList<FileStatus>(0);
+  }
+
+  public static FileStatus toFileStatus(Map<?, ?> json) {
+    Path path = new Path(getString(json, "path", ""));
+    long length = getLong(json, "length", 0);
+    boolean isdir = getBoolean(json, "isdir", false);
+    short replication = (short) getInt(json, "block_replication", -1);
+    long blockSize = getLong(json, "blocksize", 256);
+    long modificationTime = getLong(json, "modification_time", 0);
+    long accessTime = getLong(json, "access_time", 0);
+    String permString = getString(json, "permission", null);
+    FsPermission permission = toFsPermission(permString);
+    String owner = getString(json, "owner", null);
+    String group = getString(json, "group", null);
+    if (json.get("symlink") != null) {
+      Path symlink = new Path((String) json.get("symlink"));
+      return new FileStatus(length, isdir, replication, blockSize, modificationTime,
+          accessTime, permission, owner, group, symlink, path);
+    }
+    return new FileStatus(length, isdir, replication, blockSize, modificationTime,
+        accessTime, permission, owner, group, path);
+  }
+
   private static List<SnapshotDiffReport.DiffReportEntry> toDiffList(
       List<?> objs) {
     if (objs == null) {

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/web/WebHdfsFileSystem.java

@@ -1956,6 +1956,26 @@ String decodeResponse(Map<?, ?> json) throws IOException {
     }
   }
 
+  public Collection<FileStatus> getTrashRoots(boolean allUsers) {
+    statistics.incrementReadOps(1);
+    storageStatistics.incrementOpCounter(OpType.GET_TRASH_ROOTS);
+
+    final HttpOpParam.Op op = GetOpParam.Op.GETTRASHROOTS;
+    try {
+      Collection<FileStatus> trashRoots =
+          new FsPathResponseRunner<Collection<FileStatus>>(op, null,
+              new AllUsersParam(allUsers)) {
+            @Override
+            Collection<FileStatus> decodeResponse(Map<?, ?> json) throws IOException {
+              return JsonUtilClient.getTrashRoots(json);
+            }
+          }.run();
+      return trashRoots;
+    } catch (IOException e) {
+      return super.getTrashRoots(allUsers);
+    }
+  }
+
   @Override
   public void access(final Path path, final FsAction mode) throws IOException {
     final HttpOpParam.Op op = GetOpParam.Op.CHECKACCESS;

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/web/resources/AllUsersParam.java

@@ -0,0 +1,49 @@
+/**
+ * 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.
+ */
+package org.apache.hadoop.hdfs.web.resources;
+
+/** AllUsers parameter. */
+public class AllUsersParam extends BooleanParam {
+  /** Parameter name. */
+  public static final String NAME = "allusers";
+  /** Default parameter value. */
+  public static final String DEFAULT = FALSE;
+
+  private static final Domain DOMAIN = new Domain(NAME);
+
+  /**
+   * Constructor.
+   * @param value the parameter value.
+   */
+  public AllUsersParam(final Boolean value) {
+    super(DOMAIN, value);
+  }
+
+  /**
+   * Constructor.
+   * @param str a string representation of the parameter value.
+   */
+  public AllUsersParam(final String str) {
+    this(DOMAIN.parse(str));
+  }
+
+  @Override
+  public String getName() {
+    return NAME;
+  }
+}

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/web/resources/GetOpParam.java

@@ -69,6 +69,7 @@ public enum Op implements HttpOpParam.Op {
     GETSTATUS(false, HttpURLConnection.HTTP_OK),
     GETECPOLICIES(false, HttpURLConnection.HTTP_OK),
     GETECCODECS(false, HttpURLConnection.HTTP_OK),
+    GETTRASHROOTS(false, HttpURLConnection.HTTP_OK),
     GETSNAPSHOTLIST(false, HttpURLConnection.HTTP_OK);
 
     final boolean redirect;