Merge branch 'apache:trunk' into HADOOP-19136

Author: slfan1989

hadoop-common-project/hadoop-auth/pom.xml

@@ -136,7 +136,11 @@
     </dependency>
     <dependency>
       <groupId>org.apache.kerby</groupId>
-      <artifactId>kerb-simplekdc</artifactId>
+      <artifactId>kerb-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.kerby</groupId>
+      <artifactId>kerb-util</artifactId>
     </dependency>
     <dependency>
       <groupId>org.apache.directory.server</groupId>

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

@@ -0,0 +1,45 @@
+/*
+ * 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.security;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * SASL related constants.
+ */
+@InterfaceAudience.LimitedPrivate({"HDFS", "MapReduce"})
+@InterfaceStability.Evolving
+public class SaslConstants {
+  public static final Logger LOG = LoggerFactory.getLogger(SaslConstants.class);
+
+  private static final String SASL_MECHANISM_ENV = "HADOOP_SASL_MECHANISM";
+  public static final String SASL_MECHANISM;
+  private static final String SASL_MECHANISM_DEFAULT = "DIGEST-MD5";
+
+  static {
+    final String mechanism = System.getenv(SASL_MECHANISM_ENV);
+    LOG.debug("{} = {} (env)", SASL_MECHANISM_ENV, mechanism);
+    SASL_MECHANISM = mechanism != null? mechanism : SASL_MECHANISM_DEFAULT;
+    LOG.debug("{} = {} (effective)", SASL_MECHANISM_ENV, SASL_MECHANISM);
+  }
+
+  private SaslConstants() {}
+}

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

@@ -223,8 +223,8 @@ public enum AuthMethod {
     SIMPLE((byte) 80, ""),
     KERBEROS((byte) 81, "GSSAPI"),
     @Deprecated
-    DIGEST((byte) 82, "DIGEST-MD5"),
-    TOKEN((byte) 82, "DIGEST-MD5"),
+    DIGEST((byte) 82, SaslConstants.SASL_MECHANISM),
+    TOKEN((byte) 82, SaslConstants.SASL_MECHANISM),
     PLAIN((byte) 83, "PLAIN");
 
     /** The code for this method. */
@@ -273,7 +273,7 @@ public void write(DataOutput out) throws IOException {
     }
   };
 
-  /** CallbackHandler for SASL DIGEST-MD5 mechanism */
+  /** CallbackHandler for SASL mechanism. */
   @InterfaceStability.Evolving
   public static class SaslDigestCallbackHandler implements CallbackHandler {
     private SecretManager<TokenIdentifier> secretManager;
@@ -309,7 +309,7 @@ public void handle(Callback[] callbacks) throws InvalidToken,
           continue; // realm is ignored
         } else {
           throw new UnsupportedCallbackException(callback,
-              "Unrecognized SASL DIGEST-MD5 Callback");
+              "Unrecognized SASL Callback");
         }
       }
       if (pc != null) {
@@ -319,11 +319,8 @@ public void handle(Callback[] callbacks) throws InvalidToken,
         UserGroupInformation user = null;
         user = tokenIdentifier.getUser(); // may throw exception
         connection.attemptingUser = user;
-        
-        if (LOG.isDebugEnabled()) {
-          LOG.debug("SASL server DIGEST-MD5 callback: setting password "
-              + "for client: " + tokenIdentifier.getUser());
-        }
+
+        LOG.debug("SASL server callback: setting password for client: {}", user);
         pc.setPassword(password);
       }
       if (ac != null) {
@@ -339,8 +336,7 @@ public void handle(Callback[] callbacks) throws InvalidToken,
             UserGroupInformation logUser =
               getIdentifier(authzid, secretManager).getUser();
             String username = logUser == null ? null : logUser.getUserName();
-            LOG.debug("SASL server DIGEST-MD5 callback: setting "
-                + "canonicalized client ID: " + username);
+            LOG.debug("SASL server callback: setting authorizedID: {}", username);
           }
           ac.setAuthorizedID(authzid);
         }

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

@@ -326,6 +326,15 @@ Each metrics record contains tags such as HAState and Hostname as additional inf
 | `FSN(Read/Write)LockOverallNanosAvgTime` | Average time of holding the lock by all operations in nanoseconds |
 | `PendingSPSPaths` | The number of paths to be processed by storage policy satisfier |
 
+BlockManager
+-------------
+
+The metrics present statistics from the BlockManager's perspective.
+
+| Name | Description                                                                                                                     |
+|:---- |:--------------------------------------------------------------------------------------------------------------------------------|
+| `StorageTypeStats` | key represents different StorageTypes, and value represents the detailed storage information corresponding to each StorageType. |
+
 JournalNode
 -----------
 

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestSaslRPC.java

@@ -536,7 +536,7 @@ public void handle(Callback[] callbacks)
   private static Pattern BadToken =
       Pattern.compile("^" + RemoteException.class.getName() +
           "\\("+ SaslException.class.getName() + "\\): " +
-          "DIGEST-MD5: digest response format violation.*");
+          SaslConstants.SASL_MECHANISM + ": digest response format violation.*");
   private static Pattern KrbFailed =
       Pattern.compile(".*Failed on local exception:.* " +
                       "Failed to specify server's Kerberos principal name.*");

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/client/impl/metrics/BlockReaderIoProvider.java

@@ -65,7 +65,7 @@ public BlockReaderIoProvider(@Nullable ShortCircuitConf conf,
   public int read(FileChannel dataIn, ByteBuffer dst, long position)
       throws IOException{
     final int nRead;
-    if (isEnabled && (ThreadLocalRandom.current().nextInt() < sampleRangeMax)) {
+    if (isEnabled && (ThreadLocalRandom.current().nextInt(Integer.MAX_VALUE) < sampleRangeMax)) {
       long begin = timer.monotonicNow();
       nRead = dataIn.read(dst, position);
       long latency = timer.monotonicNow() - begin;

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/sasl/SaslParticipant.java

@@ -32,6 +32,7 @@
 import org.apache.hadoop.hdfs.protocol.datatransfer.IOStreamPair;
 import org.apache.hadoop.security.FastSaslClientFactory;
 import org.apache.hadoop.security.FastSaslServerFactory;
+import org.apache.hadoop.security.SaslConstants;
 import org.apache.hadoop.security.SaslInputStream;
 import org.apache.hadoop.security.SaslOutputStream;
 
@@ -50,7 +51,7 @@ class SaslParticipant {
   // a short string.
   private static final String SERVER_NAME = "0";
   private static final String PROTOCOL = "hdfs";
-  private static final String MECHANISM = "DIGEST-MD5";
+  private static final String[] MECHANISM_ARRAY = {SaslConstants.SASL_MECHANISM};
 
   // One of these will always be null.
   private final SaslServer saslServer;
@@ -81,7 +82,7 @@ public static SaslParticipant createServerSaslParticipant(
       Map<String, String> saslProps, CallbackHandler callbackHandler)
       throws SaslException {
     initializeSaslServerFactory();
-    return new SaslParticipant(saslServerFactory.createSaslServer(MECHANISM,
+    return new SaslParticipant(saslServerFactory.createSaslServer(MECHANISM_ARRAY[0],
       PROTOCOL, SERVER_NAME, saslProps, callbackHandler));
   }
 
@@ -99,7 +100,7 @@ public static SaslParticipant createClientSaslParticipant(String userName,
       throws SaslException {
     initializeSaslClientFactory();
     return new SaslParticipant(
-        saslClientFactory.createSaslClient(new String[] {MECHANISM}, userName,
+        saslClientFactory.createSaslClient(MECHANISM_ARRAY, userName,
             PROTOCOL, SERVER_NAME, saslProps, callbackHandler));
   }
 

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

@@ -189,7 +189,7 @@ public void overrideExpiredRecords(QueryResult<R> query) throws IOException {
           LOG.warn("Couldn't delete State Store record {}: {}", recordName,
               record);
         }
-      } else if (record.checkExpired(currentDriverTime)) {
+      } else if (!record.isExpired() && record.checkExpired(currentDriverTime)) {
         String recordName = StateStoreUtils.getRecordName(record.getClass());
         LOG.info("Override State Store record {}: {}", recordName, record);
         commitRecords.add(record);

hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/store/TestStateStoreMembershipState.java

@@ -29,11 +29,18 @@
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.spy;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
@@ -49,16 +56,22 @@
 import org.apache.hadoop.hdfs.server.federation.store.protocol.UpdateNamenodeRegistrationRequest;
 import org.apache.hadoop.hdfs.server.federation.store.records.MembershipState;
 import org.apache.hadoop.test.GenericTestUtils;
+import org.apache.hadoop.test.GenericTestUtils.DelayAnswer;
 import org.apache.hadoop.util.Time;
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Test the basic {@link MembershipStore} membership functionality.
  */
 public class TestStateStoreMembershipState extends TestStateStoreBase {
 
+  private static Logger LOG = LoggerFactory.getLogger(
+      TestStateStoreMembershipState.class);
+
   private static MembershipStore membershipStore;
 
   @BeforeClass
@@ -529,6 +542,94 @@ public void testRegistrationExpiredAndDeletion()
     }, 100, 3000);
   }
 
+  @Test
+  public void testRegistrationExpiredRaceCondition()
+      throws InterruptedException, IOException, TimeoutException, ExecutionException {
+
+    // Populate the state store with a single NN element
+    // 1) ns0:nn0 - Expired
+    // Create a thread to refresh the cached records, pulling the expired record
+    // into the thread's memory
+    // Then insert an active record, and confirm that the refresh thread does not
+    // override the active record with the expired record it has in memory
+
+    MembershipState.setDeletionMs(-1);
+
+    MembershipState expiredReport = createRegistration(
+        NAMESERVICES[0], NAMENODES[0], ROUTERS[0],
+        FederationNamenodeServiceState.ACTIVE);
+    expiredReport.setDateModified(Time.monotonicNow() - 5000);
+    expiredReport.setState(FederationNamenodeServiceState.EXPIRED);
+    assertTrue(namenodeHeartbeat(expiredReport));
+
+    // Load cache
+    MembershipStore memStoreSpy = spy(membershipStore);
+    DelayAnswer delayer = new DelayAnswer(LOG);
+    doAnswer(delayer).when(memStoreSpy).overrideExpiredRecords(any());
+
+    ExecutorService pool = Executors.newFixedThreadPool(1);
+
+    Future<Boolean> cacheRefreshFuture = pool.submit(() -> {
+      try {
+        return memStoreSpy.loadCache(true);
+      } catch (IOException e) {
+        LOG.error("Exception while loading cache:", e);
+      }
+      return false;
+    });
+
+    // Verify quorum and entry
+    MembershipState quorumEntry = getNamenodeRegistration(
+        expiredReport.getNameserviceId(), expiredReport.getNamenodeId());
+    assertNull(quorumEntry);
+
+
+    MembershipState record = membershipStore.getDriver()
+        .get(MembershipState.class).getRecords().get(0);
+    assertNotNull(record);
+    assertEquals(ROUTERS[0], record.getRouterId());
+    assertEquals(FederationNamenodeServiceState.EXPIRED,
+        record.getState());
+
+    // Insert active while the other thread refreshing it's cache
+    MembershipState activeReport = createRegistration(
+        NAMESERVICES[0], NAMENODES[0], ROUTERS[0],
+        FederationNamenodeServiceState.ACTIVE);
+
+    delayer.waitForCall();
+    assertTrue(namenodeHeartbeat(activeReport));
+
+    record = membershipStore.getDriver()
+        .get(MembershipState.class).getRecords().get(0);
+    assertNotNull(record);
+    assertEquals(ROUTERS[0], record.getRouterId());
+    assertEquals(FederationNamenodeServiceState.ACTIVE,
+        record.getState());
+
+    quorumEntry = getExpiredNamenodeRegistration(
+        expiredReport.getNameserviceId(), expiredReport.getNamenodeId());
+    assertNull(quorumEntry);
+
+    // Allow the thread to finish refreshing the cache
+    delayer.proceed();
+    assertTrue(cacheRefreshFuture.get(5, TimeUnit.SECONDS));
+
+    // The state store should still be the active report
+    record = membershipStore.getDriver()
+        .get(MembershipState.class).getRecords().get(0);
+    assertNotNull(record);
+    assertEquals(ROUTERS[0], record.getRouterId());
+    assertEquals(FederationNamenodeServiceState.ACTIVE,
+        record.getState());
+
+    membershipStore.loadCache(true);
+
+    quorumEntry = getExpiredNamenodeRegistration(
+        expiredReport.getNameserviceId(),
+        expiredReport.getNamenodeId());
+    assertNull(quorumEntry);
+  }
+
   @Test
   public void testNamespaceInfoWithUnavailableNameNodeRegistration()
       throws IOException {

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/sasl/SaslDataTransferServer.java

@@ -241,7 +241,7 @@ public void handle(Callback[] callbacks) throws IOException,
           continue; // realm is ignored
         } else {
           throw new UnsupportedCallbackException(callback,
-              "Unrecognized SASL DIGEST-MD5 Callback: " + callback);
+              "Unrecognized SASL Callback: " + callback);
         }
       }