HADOOP-16059. Use SASL Factories Cache to Improve Performance. Contributed by Ayush Saxena.

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

@@ -0,0 +1,80 @@
+/**
+ * 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 java.util.ArrayList;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.sasl.Sasl;
+import javax.security.sasl.SaslClient;
+import javax.security.sasl.SaslClientFactory;
+import javax.security.sasl.SaslException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+
+/**
+ * Class for dealing with caching SASL client factories.
+ */
+@InterfaceAudience.LimitedPrivate({ "HDFS", "MapReduce" })
+public class FastSaslClientFactory implements SaslClientFactory {
+  private final Map<String, List<SaslClientFactory>> factoryCache =
+      new HashMap<String, List<SaslClientFactory>>();
+
+  public FastSaslClientFactory(Map<String, ?> props) {
+    final Enumeration<SaslClientFactory> factories =
+        Sasl.getSaslClientFactories();
+    while (factories.hasMoreElements()) {
+      SaslClientFactory factory = factories.nextElement();
+      for (String mech : factory.getMechanismNames(props)) {
+        if (!factoryCache.containsKey(mech)) {
+          factoryCache.put(mech, new ArrayList<SaslClientFactory>());
+        }
+        factoryCache.get(mech).add(factory);
+      }
+    }
+  }
+
+  @Override
+  public String[] getMechanismNames(Map<String, ?> props) {
+    return factoryCache.keySet().toArray(new String[0]);
+  }
+
+  @Override
+  public SaslClient createSaslClient(String[] mechanisms,
+      String authorizationId, String protocol, String serverName,
+      Map<String, ?> props, CallbackHandler cbh) throws SaslException {
+    for (String mechanism : mechanisms) {
+      List<SaslClientFactory> factories = factoryCache.get(mechanism);
+      if (factories != null) {
+        for (SaslClientFactory factory : factories) {
+          SaslClient saslClient =
+              factory.createSaslClient(new String[] {mechanism},
+                  authorizationId, protocol, serverName, props, cbh);
+          if (saslClient != null) {
+            return saslClient;
+          }
+        }
+      }
+    }
+    return null;
+  }
+}

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

@@ -0,0 +1,78 @@
+/**
+ * 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 java.util.ArrayList;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.sasl.Sasl;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+import javax.security.sasl.SaslServerFactory;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+
+/**
+ * Class for dealing with caching SASL server factories.
+ */
+@InterfaceAudience.LimitedPrivate({"HDFS", "MapReduce"})
+public class FastSaslServerFactory implements SaslServerFactory {
+  private final Map<String, List<SaslServerFactory>> factoryCache =
+      new HashMap<String, List<SaslServerFactory>>();
+
+  public FastSaslServerFactory(Map<String, ?> props) {
+    final Enumeration<SaslServerFactory> factories =
+        Sasl.getSaslServerFactories();
+    while (factories.hasMoreElements()) {
+      SaslServerFactory factory = factories.nextElement();
+      for (String mech : factory.getMechanismNames(props)) {
+        if (!factoryCache.containsKey(mech)) {
+          factoryCache.put(mech, new ArrayList<SaslServerFactory>());
+        }
+        factoryCache.get(mech).add(factory);
+      }
+    }
+  }
+
+  @Override
+  public SaslServer createSaslServer(String mechanism, String protocol,
+      String serverName, Map<String, ?> props, CallbackHandler cbh)
+      throws SaslException {
+    SaslServer saslServer = null;
+    List<SaslServerFactory> factories = factoryCache.get(mechanism);
+    if (factories != null) {
+      for (SaslServerFactory factory : factories) {
+        saslServer = factory.createSaslServer(
+            mechanism, protocol, serverName, props, cbh);
+        if (saslServer != null) {
+          break;
+        }
+      }
+    }
+    return saslServer;
+  }
+
+  @Override
+  public String[] getMechanismNames(Map<String, ?> props) {
+    return factoryCache.keySet().toArray(new String[0]);
+  }
+}

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

@@ -44,6 +44,7 @@
 import javax.security.sasl.Sasl;
 import javax.security.sasl.SaslException;
 import javax.security.sasl.SaslClient;
+import javax.security.sasl.SaslClientFactory;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
@@ -93,6 +94,7 @@ public class SaslRpcClient {
   private SaslClient saslClient;
   private SaslPropertiesResolver saslPropsResolver;
   private AuthMethod authMethod;
+  private static SaslClientFactory saslFactory;
   
   private static final RpcRequestHeaderProto saslHeader = ProtoUtil
       .makeRpcRequestHeader(RpcKind.RPC_PROTOCOL_BUFFER,
@@ -101,6 +103,10 @@ public class SaslRpcClient {
   private static final RpcSaslProto negotiateRequest =
       RpcSaslProto.newBuilder().setState(SaslState.NEGOTIATE).build();
 
+  static {
+    saslFactory = new FastSaslClientFactory(null);
+  }
+
   /**
    * Create a SaslRpcClient that can be used by a RPC client to negotiate
    * SASL authentication with a RPC server
@@ -251,7 +257,7 @@ private SaslClient createSaslClient(SaslAuth authType)
       LOG.debug("Creating SASL " + mechanism + "(" + method + ") "
           + " client to authenticate to service at " + saslServerName);
     }
-    return Sasl.createSaslClient(
+    return saslFactory.createSaslClient(
         new String[] {mechanism}, saslUser, saslProtocol, saslServerName,
         saslProperties, saslCallback);
   }

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

@@ -26,10 +26,6 @@
 import java.nio.charset.StandardCharsets;
 import java.security.PrivilegedExceptionAction;
 import java.security.Security;
-import java.util.ArrayList;
-import java.util.Enumeration;
-import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 
 import javax.security.auth.callback.Callback;
@@ -39,7 +35,6 @@
 import javax.security.auth.callback.UnsupportedCallbackException;
 import javax.security.sasl.AuthorizeCallback;
 import javax.security.sasl.RealmCallback;
-import javax.security.sasl.Sasl;
 import javax.security.sasl.SaslException;
 import javax.security.sasl.SaslServer;
 import javax.security.sasl.SaslServerFactory;
@@ -178,11 +173,13 @@ public SaslServer run() throws SaslException  {
   }
 
   public static void init(Configuration conf) {
+    if (saslFactory == null) {
       Security.addProvider(new SaslPlainServer.SecurityProvider());
       // passing null so factory is populated with all possibilities. the
       // properties passed when instantiating a server are what really matter
       saslFactory = new FastSaslServerFactory(null);
     }
+  }
   
   static String encodeIdentifier(byte[] identifier) {
     return new String(Base64.encodeBase64(identifier), StandardCharsets.UTF_8);
@@ -367,47 +364,4 @@ public void handle(Callback[] callbacks) throws
       }
     }
   }
-  
-  // Sasl.createSaslServer is 100-200X slower than caching the factories!
-  private static class FastSaslServerFactory implements SaslServerFactory {
-    private final Map<String,List<SaslServerFactory>> factoryCache =
-        new HashMap<String,List<SaslServerFactory>>();
-
-    FastSaslServerFactory(Map<String,?> props) {
-      final Enumeration<SaslServerFactory> factories =
-          Sasl.getSaslServerFactories();
-      while (factories.hasMoreElements()) {
-        SaslServerFactory factory = factories.nextElement();
-        for (String mech : factory.getMechanismNames(props)) {
-          if (!factoryCache.containsKey(mech)) {
-            factoryCache.put(mech, new ArrayList<SaslServerFactory>());
-          }
-          factoryCache.get(mech).add(factory);
-        }
-      }
-    }
-
-    @Override
-    public SaslServer createSaslServer(String mechanism, String protocol,
-        String serverName, Map<String,?> props, CallbackHandler cbh)
-        throws SaslException {
-      SaslServer saslServer = null;
-      List<SaslServerFactory> factories = factoryCache.get(mechanism);
-      if (factories != null) {
-        for (SaslServerFactory factory : factories) {
-          saslServer = factory.createSaslServer(
-              mechanism, protocol, serverName, props, cbh);
-          if (saslServer != null) {
-            break;
-          }
-        }
-      }
-      return saslServer;
-    }
-
-    @Override
-    public String[] getMechanismNames(Map<String, ?> props) {
-      return factoryCache.keySet().toArray(new String[0]);
-    }
-  }
 }

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

@@ -23,11 +23,15 @@
 import javax.security.auth.callback.CallbackHandler;
 import javax.security.sasl.Sasl;
 import javax.security.sasl.SaslClient;
+import javax.security.sasl.SaslClientFactory;
 import javax.security.sasl.SaslException;
 import javax.security.sasl.SaslServer;
+import javax.security.sasl.SaslServerFactory;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 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.SaslInputStream;
 import org.apache.hadoop.security.SaslOutputStream;
 
@@ -51,7 +55,20 @@ class SaslParticipant {
   // One of these will always be null.
   private final SaslServer saslServer;
   private final SaslClient saslClient;
+  private static SaslServerFactory saslServerFactory;
+  private static SaslClientFactory saslClientFactory;
 
+  private static void initializeSaslServerFactory() {
+    if (saslServerFactory == null) {
+      saslServerFactory = new FastSaslServerFactory(null);
+    }
+  }
+
+  private static void initializeSaslClientFactory() {
+    if (saslClientFactory == null) {
+      saslClientFactory = new FastSaslClientFactory(null);
+    }
+  }
   /**
    * Creates a SaslParticipant wrapping a SaslServer.
    *
@@ -63,7 +80,8 @@ class SaslParticipant {
   public static SaslParticipant createServerSaslParticipant(
       Map<String, String> saslProps, CallbackHandler callbackHandler)
       throws SaslException {
-    return new SaslParticipant(Sasl.createSaslServer(MECHANISM,
+    initializeSaslServerFactory();
+    return new SaslParticipant(saslServerFactory.createSaslServer(MECHANISM,
       PROTOCOL, SERVER_NAME, saslProps, callbackHandler));
   }
 
@@ -79,8 +97,10 @@ public static SaslParticipant createServerSaslParticipant(
   public static SaslParticipant createClientSaslParticipant(String userName,
       Map<String, String> saslProps, CallbackHandler callbackHandler)
       throws SaslException {
-    return new SaslParticipant(Sasl.createSaslClient(new String[] { MECHANISM },
+    initializeSaslClientFactory();
-      userName, PROTOCOL, SERVER_NAME, saslProps, callbackHandler));
+    return new SaslParticipant(
+        saslClientFactory.createSaslClient(new String[] {MECHANISM}, userName,
+            PROTOCOL, SERVER_NAME, saslProps, callbackHandler));
   }
 
   /**