Add a new conf to set SASL mechanism.

Author: szetszwo

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeysPublic.java

@@ -736,6 +736,10 @@ public class CommonConfigurationKeysPublic {
    */
   public static final String  HADOOP_RPC_PROTECTION =
     "hadoop.rpc.protection";
+  public static final String HADOOP_SECURITY_SASL_MECHANISM_KEY
+      = "hadoop.security.sasl.mechanism";
+  public static final String HADOOP_SECURITY_SASL_MECHANISM_DEFAULT
+      = "DIGEST-MD5";
   public static final String HADOOP_SECURITY_SASL_CUSTOMIZEDCALLBACKHANDLER_CLASS_KEY
       = "hadoop.security.sasl.CustomizedCallbackHandler.class";
   /** Class to override Sasl Properties for a connection */

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

@@ -2610,7 +2610,7 @@ private RpcSaslProto buildSaslNegotiateResponse()
       // accelerate token negotiation by sending initial challenge
       // in the negotiation response
       if (enabledAuthMethods.contains(AuthMethod.TOKEN)
-          && SaslConstants.SASL_MECHANISM_DEFAULT.equals(AuthMethod.TOKEN.getMechanismName())) {
+          && SaslConstants.isDefaultMechanism(AuthMethod.TOKEN.getMechanismName())) {
         saslServer = createSaslServer(AuthMethod.TOKEN);
         byte[] challenge = saslServer.evaluateResponse(new byte[0]);
         RpcSaslProto.Builder negotiateBuilder =

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

@@ -48,11 +48,17 @@ private static synchronized CustomizedCallbackHandler getSynchronously(
       //cache miss
       final Class<?> clazz = conf.getClass(key, DefaultHandler.class);
       LOG.info("{} = {}", key, clazz);
+      if (clazz == DefaultHandler.class) {
+        return DefaultHandler.INSTANCE;
+      }
+
       final Object created;
       try {
         created = clazz.newInstance();
       } catch (Exception e) {
-        throw new IllegalStateException("Failed to create a new instance of " + clazz, e);
+        LOG.warn("Failed to create a new instance of {}, fallback to {}",
+            clazz, DefaultHandler.class, e);
+        return DefaultHandler.INSTANCE;
       }
 
       final CustomizedCallbackHandler handler = created instanceof CustomizedCallbackHandler ?
@@ -74,6 +80,8 @@ private Cache() { }
   }
 
   class DefaultHandler implements CustomizedCallbackHandler {
+    private static final DefaultHandler INSTANCE = new DefaultHandler();
+
     @Override
     public void handleCallbacks(List<Callback> callbacks, String username, char[] password)
         throws UnsupportedCallbackException {

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

@@ -19,9 +19,13 @@
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.conf.Configuration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_SASL_MECHANISM_DEFAULT;
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_SASL_MECHANISM_KEY;
+
 /**
  * SASL related constants.
  */
@@ -32,14 +36,25 @@ public class SaslConstants {
 
   private static final String SASL_MECHANISM_ENV = "HADOOP_SASL_MECHANISM";
   public static final String SASL_MECHANISM;
-  public static final String SASL_MECHANISM_DEFAULT = "DIGEST-MD5";
 
   static {
-    final String mechanism = System.getenv(SASL_MECHANISM_ENV);
+    // env
+    String mechanism = System.getenv(SASL_MECHANISM_ENV);
     LOG.debug("{} = {} (env)", SASL_MECHANISM_ENV, mechanism);
-    SASL_MECHANISM = mechanism != null? mechanism : SASL_MECHANISM_DEFAULT;
+
+    if (mechanism == null) {
+      // conf
+      final Configuration conf = new Configuration();
+      mechanism = conf.get(HADOOP_SECURITY_SASL_MECHANISM_KEY, HADOOP_SECURITY_SASL_MECHANISM_DEFAULT);
+    }
+
+    SASL_MECHANISM = mechanism != null? mechanism : HADOOP_SECURITY_SASL_MECHANISM_DEFAULT;
     LOG.debug("{} = {} (effective)", SASL_MECHANISM_ENV, SASL_MECHANISM);
   }
 
+  public static boolean isDefaultMechanism(String mechanism) {
+    return HADOOP_SECURITY_SASL_MECHANISM_DEFAULT.equals(mechanism);
+  }
+
   private SaslConstants() {}
 }

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

@@ -236,6 +236,8 @@ public enum AuthMethod {
     private AuthMethod(byte code, String mechanismName) { 
       this.code = code;
       this.mechanismName = mechanismName;
+      LOG.info("{} {}: code={}, mechanism=\"{}\"",
+          getClass().getSimpleName(), name(), code, mechanismName);
     }
 
     private static final int FIRST_CODE = values()[0].code;

hadoop-common-project/hadoop-common/src/main/resources/core-default.xml

@@ -732,6 +732,14 @@
   </description>
 </property>
 
+<property>
+  <name>hadoop.security.sasl.mechanism</name>
+  <value>DIGEST-MD5</value>
+  <description>
+    The SASL mechanism used in Hadoop.
+  </description>
+</property>
+
 <property>
   <name>hadoop.security.sasl.CustomizedCallbackHandler.class</name>
   <value></value>

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

@@ -127,7 +127,7 @@ private SaslParticipant(SaslClient saslClient) {
   }
 
   byte[] createFirstMessage() throws SaslException {
-    return MECHANISM_ARRAY[0].equals(SaslConstants.SASL_MECHANISM_DEFAULT) ? EMPTY_BYTE_ARRAY
+    return SaslConstants.isDefaultMechanism(MECHANISM_ARRAY[0]) ? EMPTY_BYTE_ARRAY
         : evaluateChallengeOrResponse(EMPTY_BYTE_ARRAY);
   }