Merge branch 'apache:trunk' into YARN-7953

Author: slfan1989

LICENSE-binary

@@ -337,7 +337,7 @@ org.apache.kerby:kerby-xdr:2.0.3
 org.apache.kerby:token-provider:2.0.3
 org.apache.solr:solr-solrj:8.11.2
 org.apache.yetus:audience-annotations:0.5.0
-org.apache.zookeeper:zookeeper:3.7.2
+org.apache.zookeeper:zookeeper:3.8.3
 org.codehaus.jettison:jettison:1.5.4
 org.eclipse.jetty:jetty-annotations:9.4.53.v20231009
 org.eclipse.jetty:jetty-http:9.4.53.v20231009

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/alias/CredentialShell.java

@@ -25,6 +25,7 @@
 import java.util.Arrays;
 import java.util.List;
 
+import org.apache.hadoop.security.alias.CredentialProvider.CredentialEntry;
 import org.apache.hadoop.classification.VisibleForTesting;
 
 import org.apache.commons.lang3.StringUtils;
@@ -365,12 +366,17 @@ public void execute() throws IOException, NoSuchAlgorithmException {
         } else {
           password = c.readPassword("Enter alias password: ");
         }
-        char[] storePassword =
-            provider.getCredentialEntry(alias).getCredential();
-        String beMatch =
-            Arrays.equals(storePassword, password) ? "success" : "failed";
+        CredentialEntry credentialEntry = provider.getCredentialEntry(alias);
+        if(credentialEntry == null) {
+          // Fail the password match when alias not found
+          getOut().println("Password match failed for " + alias + ".");
+        } else {
+          char[] storePassword = credentialEntry.getCredential();
+          String beMatch =
+              Arrays.equals(storePassword, password) ? "success" : "failed";
 
-        getOut().println("Password match " + beMatch + " for " +  alias + ".");
+          getOut().println("Password match " + beMatch + " for " + alias + ".");
+        }
       } catch (IOException e) {
         getOut().println("Cannot check aliases for CredentialProvider: " +
             provider.toString()

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/alias/TestCredShell.java

@@ -165,6 +165,21 @@ public void testPromptForCredentialWithEmptyPasswd() throws Exception {
     assertTrue(outContent.toString().contains("Passwords don't match"));
   }
 
+  @Test
+  public void testPromptForCredentialNotFound() throws Exception {
+    String[] args1 = {"check", "credential1", "-provider",
+        jceksProvider};
+    ArrayList<String> password = new ArrayList<String>();
+    password.add("p@ssw0rd");
+    int rc = 0;
+    CredentialShell shell = new CredentialShell();
+    shell.setConf(new Configuration());
+    shell.setPasswordReader(new MockPasswordReader(password));
+    rc = shell.run(args1);
+    assertEquals(0, rc);
+    assertOutputContains("Password match failed for credential1.");
+  }
+
   @Test
   public void testPromptForCredential() throws Exception {
     String[] args1 = {"create", "credential1", "-provider",

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

@@ -787,7 +787,7 @@ public void run() {
             scope = null;
             dataQueue.removeFirst();
             ackQueue.addLast(one);
-            packetSendTime.put(one.getSeqno(), Time.monotonicNow());
+            packetSendTime.put(one.getSeqno(), Time.monotonicNowNanos());
             dataQueue.notifyAll();
           }
         }
@@ -953,7 +953,7 @@ void waitForAckedSeqno(long seqno) throws IOException {
         dnodes = nodes != null ? nodes.length : 3;
       }
       int writeTimeout = dfsClient.getDatanodeWriteTimeout(dnodes);
-      long begin = Time.monotonicNow();
+      long begin = Time.monotonicNowNanos();
       try {
         synchronized (dataQueue) {
           while (!streamerClosed) {
@@ -963,14 +963,14 @@ void waitForAckedSeqno(long seqno) throws IOException {
             }
             try {
               dataQueue.wait(1000); // when we receive an ack, we notify on
-              long duration = Time.monotonicNow() - begin;
-              if (duration > writeTimeout) {
+              long duration = Time.monotonicNowNanos() - begin;
+              if (TimeUnit.NANOSECONDS.toMillis(duration) > writeTimeout) {
                 LOG.error("No ack received, took {}ms (threshold={}ms). "
                     + "File being written: {}, block: {}, "
                     + "Write pipeline datanodes: {}.",
-                    duration, writeTimeout, src, block, nodes);
+                    TimeUnit.NANOSECONDS.toMillis(duration), writeTimeout, src, block, nodes);
                 throw new InterruptedIOException("No ack received after " +
-                    duration / 1000 + "s and a timeout of " +
+                    TimeUnit.NANOSECONDS.toSeconds(duration) + "s and a timeout of " +
                     writeTimeout / 1000 + "s");
               }
               // dataQueue
@@ -984,11 +984,12 @@ void waitForAckedSeqno(long seqno) throws IOException {
       } catch (ClosedChannelException cce) {
         LOG.debug("Closed channel exception", cce);
       }
-      long duration = Time.monotonicNow() - begin;
-      if (duration > dfsclientSlowLogThresholdMs) {
+      long duration = Time.monotonicNowNanos() - begin;
+      if (TimeUnit.NANOSECONDS.toMillis(duration) > dfsclientSlowLogThresholdMs) {
         LOG.warn("Slow waitForAckedSeqno took {}ms (threshold={}ms). File being"
                 + " written: {}, block: {}, Write pipeline datanodes: {}.",
-            duration, dfsclientSlowLogThresholdMs, src, block, nodes);
+            TimeUnit.NANOSECONDS.toMillis(duration), dfsclientSlowLogThresholdMs,
+            src, block, nodes);
       }
     }
   }
@@ -1179,10 +1180,10 @@ public void run() {
           if (ack.getSeqno() != DFSPacket.HEART_BEAT_SEQNO) {
             Long begin = packetSendTime.get(ack.getSeqno());
             if (begin != null) {
-              long duration = Time.monotonicNow() - begin;
-              if (duration > dfsclientSlowLogThresholdMs) {
+              long duration = Time.monotonicNowNanos() - begin;
+              if (TimeUnit.NANOSECONDS.toMillis(duration) > dfsclientSlowLogThresholdMs) {
                 LOG.info("Slow ReadProcessor read fields for block " + block
-                    + " took " + duration + "ms (threshold="
+                    + " took " + TimeUnit.NANOSECONDS.toMillis(duration) + "ms (threshold="
                     + dfsclientSlowLogThresholdMs + "ms); ack: " + ack
                     + ", targets: " + Arrays.asList(targets));
               }

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/Quota.java

@@ -75,7 +75,7 @@ public Quota(Router router, RouterRpcServer server) {
    * @param storagespaceQuota Storage space quota.
    * @param type StorageType that the space quota is intended to be set on.
    * @param checkMountEntry whether to check the path is a mount entry.
-   * @throws AccessControlException If the quota system is disabled or if
+   * @throws IOException If the quota system is disabled or if
    * checkMountEntry is true and the path is a mount entry.
    */
   public void setQuota(String path, long namespaceQuota, long storagespaceQuota,

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RBFConfigKeys.java

@@ -279,6 +279,10 @@ public class RBFConfigKeys extends CommonConfigurationKeysPublic {
       FEDERATION_ROUTER_PREFIX + "safemode.expiration";
   public static final long DFS_ROUTER_SAFEMODE_EXPIRATION_DEFAULT =
       3 * DFS_ROUTER_CACHE_TIME_TO_LIVE_MS_DEFAULT;
+  public static final String DFS_ROUTER_SAFEMODE_CHECKPERIOD_MS =
+      FEDERATION_ROUTER_PREFIX + "safemode.checkperiod";
+  public static final long DFS_ROUTER_SAFEMODE_CHECKPERIOD_MS_DEFAULT =
+      TimeUnit.SECONDS.toMillis(5);
 
   // HDFS Router-based federation mount table entries
   /** Maximum number of cache entries to have. */

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterSafemodeService.java

@@ -133,8 +133,8 @@ protected void serviceInit(Configuration conf) throws Exception {
 
     // Use same interval as cache update service
     this.setIntervalMs(conf.getTimeDuration(
-        RBFConfigKeys.DFS_ROUTER_CACHE_TIME_TO_LIVE_MS,
-        RBFConfigKeys.DFS_ROUTER_CACHE_TIME_TO_LIVE_MS_DEFAULT,
+        RBFConfigKeys.DFS_ROUTER_SAFEMODE_CHECKPERIOD_MS,
+        RBFConfigKeys.DFS_ROUTER_SAFEMODE_CHECKPERIOD_MS_DEFAULT,
         TimeUnit.MILLISECONDS));
 
     this.startupInterval = conf.getTimeDuration(

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/StateStoreService.java

@@ -138,7 +138,7 @@ public StateStoreService() {
    * Initialize the State Store and the connection to the back-end.
    *
    * @param config Configuration for the State Store.
-   * @throws IOException Cannot create driver for the State Store.
+   * @throws Exception Cannot create driver for the State Store.
    */
   @Override
   protected void serviceInit(Configuration config) throws Exception {
@@ -239,7 +239,6 @@ protected void serviceStop() throws Exception {
    *
    * @param <T> Type of the records stored.
    * @param clazz Class of the record store to track.
-   * @return New record store.
    * @throws ReflectiveOperationException
    */
   private <T extends RecordStore<?>> void addRecordStore(
@@ -428,7 +427,6 @@ public void refreshCaches(boolean force) {
           result = cachedStore.loadCache(force);
         } catch (IOException e) {
           LOG.error("Error updating cache for {}", cacheName, e);
-          result = false;
         }
         if (!result) {
           success = false;

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/impl/MountTableStoreImpl.java

@@ -70,7 +70,7 @@ public MountTableStoreImpl(StateStoreDriver driver) {
    *
    * @param src mount entry being accessed
    * @param action type of action being performed on the mount entry
-   * @throws AccessControlException if mount table cannot be accessed
+   * @throws IOException if mount table cannot be accessed
    */
   private void checkMountTableEntryPermission(String src, FsAction action)
       throws IOException {
@@ -90,7 +90,7 @@ private void checkMountTableEntryPermission(String src, FsAction action)
    * Check parent path permission recursively. It needs WRITE permission
    * of the nearest parent entry and other EXECUTE permission.
    * @param src mount entry being checked
-   * @throws AccessControlException if mount table cannot be accessed
+   * @throws IOException if mount table cannot be accessed
    */
   private void checkMountTablePermission(final String src) throws IOException {
     String parent = src.substring(0, src.lastIndexOf(Path.SEPARATOR));

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/store/protocol/impl/pb/FederationProtocolPBTranslator.java

@@ -54,7 +54,7 @@ public FederationProtocolPBTranslator(Class<P> protoType) {
    * the proto handler this translator holds.
    */
   @SuppressWarnings("unchecked")
-  public void setProto(Message p) {
+  public void setProto(Message p) throws IllegalArgumentException {
     if (protoClass.isInstance(p)) {
       if (this.builder != null) {
         // Merge with builder