HBASE-29144 Client request fails for KERBEROS with RpcConnectionRegis…

hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AbstractRpcClient.java

@@ -43,6 +43,7 @@
 import org.apache.hadoop.hbase.net.Address;
 import org.apache.hadoop.hbase.security.User;
 import org.apache.hadoop.hbase.security.UserProvider;
+import org.apache.hadoop.hbase.security.provider.SaslClientAuthenticationProviders;
 import org.apache.hadoop.hbase.trace.TraceUtil;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.apache.hadoop.hbase.util.PoolMap;
@@ -116,6 +117,8 @@ public abstract class AbstractRpcClient<T extends RpcConnection> implements RpcC
   protected final UserProvider userProvider;
   protected final CellBlockBuilder cellBlockBuilder;
 
+  protected final SaslClientAuthenticationProviders providers;
+
   protected final int minIdleTimeBeforeClose; // if the connection is idle for more than this
   // time (in ms), it will be closed at any moment.
   protected final int maxRetries; // the max. no. of retries for socket connections
@@ -184,6 +187,8 @@ public AbstractRpcClient(Configuration conf, String clusterId, SocketAddress loc
       conf.getInt(HConstants.HBASE_CLIENT_PERSERVER_REQUESTS_THRESHOLD,
         HConstants.DEFAULT_HBASE_CLIENT_PERSERVER_REQUESTS_THRESHOLD);
 
+    this.providers = new SaslClientAuthenticationProviders(conf);
+
     this.connections = new PoolMap<>(getPoolType(conf), getPoolSize(conf));
 
     this.cleanupIdleConnectionTask = IDLE_CONN_SWEEPER.scheduleAtFixedRate(new Runnable() {

hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/BlockingRpcConnection.java

@@ -220,8 +220,9 @@ public void cleanup(IOException e) {
 
   BlockingRpcConnection(BlockingRpcClient rpcClient, ConnectionId remoteId) throws IOException {
     super(rpcClient.conf, AbstractRpcClient.WHEEL_TIMER, remoteId, rpcClient.clusterId,
-      rpcClient.userProvider.isHBaseSecurityEnabled(), rpcClient.codec, rpcClient.compressor,
-      rpcClient.cellBlockBuilder, rpcClient.metrics, rpcClient.connectionAttributes);
+      rpcClient.userProvider.isHBaseSecurityEnabled(), rpcClient.providers, rpcClient.codec,
+      rpcClient.compressor, rpcClient.cellBlockBuilder, rpcClient.metrics,
+      rpcClient.connectionAttributes);
     this.rpcClient = rpcClient;
     this.connectionHeaderPreamble = getConnectionHeaderPreamble();
     ConnectionHeader header = getConnectionHeader();

hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcConnection.java

@@ -106,8 +106,9 @@ class NettyRpcConnection extends RpcConnection {
 
   NettyRpcConnection(NettyRpcClient rpcClient, ConnectionId remoteId) throws IOException {
     super(rpcClient.conf, AbstractRpcClient.WHEEL_TIMER, remoteId, rpcClient.clusterId,
-      rpcClient.userProvider.isHBaseSecurityEnabled(), rpcClient.codec, rpcClient.compressor,
-      rpcClient.cellBlockBuilder, rpcClient.metrics, rpcClient.connectionAttributes);
+      rpcClient.userProvider.isHBaseSecurityEnabled(), rpcClient.providers, rpcClient.codec,
+      rpcClient.compressor, rpcClient.cellBlockBuilder, rpcClient.metrics,
+      rpcClient.connectionAttributes);
     this.rpcClient = rpcClient;
     this.eventLoop = rpcClient.group.next();
     byte[] connectionHeaderPreamble = getConnectionHeaderPreamble();

hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RpcConnection.java

@@ -119,9 +119,9 @@ abstract class RpcConnection {
   private String lastSucceededServerPrincipal;
 
   protected RpcConnection(Configuration conf, HashedWheelTimer timeoutTimer, ConnectionId remoteId,
-    String clusterId, boolean isSecurityEnabled, Codec codec, CompressionCodec compressor,
-    CellBlockBuilder cellBlockBuilder, MetricsConnection metrics,
-    Map<String, byte[]> connectionAttributes) throws IOException {
+    String clusterId, boolean isSecurityEnabled, SaslClientAuthenticationProviders providers,
+    Codec codec, CompressionCodec compressor, CellBlockBuilder cellBlockBuilder,
+    MetricsConnection metrics, Map<String, byte[]> connectionAttributes) throws IOException {
     this.timeoutTimer = timeoutTimer;
     this.codec = codec;
     this.compressor = compressor;
@@ -134,8 +134,6 @@ protected RpcConnection(Configuration conf, HashedWheelTimer timeoutTimer, Conne
     this.useSasl = isSecurityEnabled;
 
     // Choose the correct Token and AuthenticationProvider for this client to use
-    SaslClientAuthenticationProviders providers =
-      SaslClientAuthenticationProviders.getInstance(conf);
     Pair<SaslClientAuthenticationProvider, Token<? extends TokenIdentifier>> pair;
     if (useSasl && securityInfo != null) {
       pair = providers.selectProvider(clusterId, ticket);

hbase-client/src/main/java/org/apache/hadoop/hbase/security/provider/AuthenticationProviderSelector.java

@@ -32,9 +32,10 @@
 public interface AuthenticationProviderSelector {
 
   /**
-   * Initializes the implementation with configuration and a set of providers available. This method
-   * should be called exactly once per implementation prior to calling
-   * {@link #selectProvider(String, User)}.
+   * Initializes the implementation with configuration and a set of providers available.
+   * <p>
+   * This method is called once upon construction, before {@link #selectProvider(String, User)} is
+   * invoked.
    */
   void configure(Configuration conf,
     Collection<SaslClientAuthenticationProvider> availableProviders);

hbase-client/src/main/java/org/apache/hadoop/hbase/security/provider/SaslClientAuthenticationProviders.java

@@ -48,16 +48,36 @@ public final class SaslClientAuthenticationProviders {
   public static final String SELECTOR_KEY = "hbase.client.sasl.provider.class";
   public static final String EXTRA_PROVIDERS_KEY = "hbase.client.sasl.provider.extras";
 
-  private static final AtomicReference<SaslClientAuthenticationProviders> providersRef =
+  private static final AtomicReference<SaslClientAuthenticationProviders> PROVIDER_REF =
     new AtomicReference<>();
 
   private final Collection<SaslClientAuthenticationProvider> providers;
   private final AuthenticationProviderSelector selector;
 
-  private SaslClientAuthenticationProviders(Collection<SaslClientAuthenticationProvider> providers,
-    AuthenticationProviderSelector selector) {
-    this.providers = providers;
-    this.selector = selector;
+  /**
+   * Creates a new instance of SaslClientAuthenticationProviders.
+   * @param conf the configuration to use for loading providers and selector
+   */
+  public SaslClientAuthenticationProviders(Configuration conf) {
+    ServiceLoader<SaslClientAuthenticationProvider> loader =
+      ServiceLoader.load(SaslClientAuthenticationProvider.class,
+        SaslClientAuthenticationProviders.class.getClassLoader());
+    HashMap<Byte, SaslClientAuthenticationProvider> providerMap = new HashMap<>();
+    for (SaslClientAuthenticationProvider provider : loader) {
+      addProviderIfNotExists(provider, providerMap);
+    }
+
+    addExplicitProviders(conf, providerMap);
+
+    providers = Collections.unmodifiableCollection(providerMap.values());
+
+    if (LOG.isTraceEnabled()) {
+      String loadedProviders = providers.stream().map((provider) -> provider.getClass().getName())
+        .collect(Collectors.joining(", "));
+      LOG.trace("Found SaslClientAuthenticationProviders {}", loadedProviders);
+    }
+
+    selector = instantiateSelector(conf, providers);
   }
 
   /**
@@ -69,22 +89,30 @@ public int getNumRegisteredProviders() {
 
   /**
    * Returns a singleton instance of {@link SaslClientAuthenticationProviders}.
+   * @deprecated Since 2.5.14 and 2.6.4, will be removed in newer minor release lines. This class
+   *             should not be singleton, please do not use it any more. see HBASE-29144 for more
+   *             details.
    */
+  @Deprecated
   public static synchronized SaslClientAuthenticationProviders getInstance(Configuration conf) {
-    SaslClientAuthenticationProviders providers = providersRef.get();
+    SaslClientAuthenticationProviders providers = PROVIDER_REF.get();
     if (providers == null) {
-      providers = instantiate(conf);
-      providersRef.set(providers);
+      providers = new SaslClientAuthenticationProviders(conf);
+      PROVIDER_REF.set(providers);
     }
 
     return providers;
   }
 
   /**
    * Removes the cached singleton instance of {@link SaslClientAuthenticationProviders}.
+   * @deprecated Since 2.5.14 and 2.6.4, will be removed in newer minor release lines. This class
+   *             should not be singleton, please do not use it any more. see HBASE-29144 for more
+   *             details.
    */
+  @Deprecated
   public static synchronized void reset() {
-    providersRef.set(null);
+    PROVIDER_REF.set(null);
   }
 
   /**
@@ -93,7 +121,7 @@ public static synchronized void reset() {
    */
   static void addProviderIfNotExists(SaslClientAuthenticationProvider provider,
     HashMap<Byte, SaslClientAuthenticationProvider> providers) {
-    Byte code = provider.getSaslAuthMethod().getCode();
+    byte code = provider.getSaslAuthMethod().getCode();
     SaslClientAuthenticationProvider existingProvider = providers.get(code);
     if (existingProvider != null) {
       throw new RuntimeException("Already registered authentication provider with " + code + " "
@@ -105,7 +133,7 @@ static void addProviderIfNotExists(SaslClientAuthenticationProvider provider,
   /**
    * Instantiates the ProviderSelector implementation from the provided configuration.
    */
-  static AuthenticationProviderSelector instantiateSelector(Configuration conf,
+  private static AuthenticationProviderSelector instantiateSelector(Configuration conf,
     Collection<SaslClientAuthenticationProvider> providers) {
     Class<? extends AuthenticationProviderSelector> clz = conf.getClass(SELECTOR_KEY,
       BuiltInProviderSelector.class, AuthenticationProviderSelector.class);
@@ -161,34 +189,6 @@ static void addExplicitProviders(Configuration conf,
     }
   }
 
-  /**
-   * Instantiates all client authentication providers and returns an instance of
-   * {@link SaslClientAuthenticationProviders}.
-   */
-  static SaslClientAuthenticationProviders instantiate(Configuration conf) {
-    ServiceLoader<SaslClientAuthenticationProvider> loader =
-      ServiceLoader.load(SaslClientAuthenticationProvider.class,
-        SaslClientAuthenticationProviders.class.getClassLoader());
-    HashMap<Byte, SaslClientAuthenticationProvider> providerMap = new HashMap<>();
-    for (SaslClientAuthenticationProvider provider : loader) {
-      addProviderIfNotExists(provider, providerMap);
-    }
-
-    addExplicitProviders(conf, providerMap);
-
-    Collection<SaslClientAuthenticationProvider> providers =
-      Collections.unmodifiableCollection(providerMap.values());
-
-    if (LOG.isTraceEnabled()) {
-      String loadedProviders = providers.stream().map((provider) -> provider.getClass().getName())
-        .collect(Collectors.joining(", "));
-      LOG.trace("Found SaslClientAuthenticationProviders {}", loadedProviders);
-    }
-
-    AuthenticationProviderSelector selector = instantiateSelector(conf, providers);
-    return new SaslClientAuthenticationProviders(providers, selector);
-  }
-
   /**
    * Returns the provider and token pair for SIMPLE authentication. This method is a "hack" while
    * SIMPLE authentication for HBase does not flow through the SASL codepath.

hbase-client/src/test/java/org/apache/hadoop/hbase/security/provider/TestSaslClientAuthenticationProviders.java

@@ -17,17 +17,16 @@
  */
 package org.apache.hadoop.hbase.security.provider;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotSame;
-import static org.junit.Assert.assertSame;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 import java.io.IOException;
 import java.net.InetAddress;
 import java.util.HashMap;
 import java.util.Map;
 import javax.security.sasl.SaslClient;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HBaseClassTestRule;
 import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.security.SecurityInfo;
 import org.apache.hadoop.hbase.security.User;
@@ -36,19 +35,15 @@
 import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.security.token.TokenIdentifier;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
 
 import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.UserInformation;
 
-@Category({ SmallTests.class, SecurityTests.class })
+@Tag(SmallTests.TAG)
+@Tag(SecurityTests.TAG)
 public class TestSaslClientAuthenticationProviders {
 
-  @ClassRule
-  public static final HBaseClassTestRule CLASS_RULE =
-    HBaseClassTestRule.forClass(TestSaslClientAuthenticationProviders.class);
-
   @Test
   public void testCannotAddTheSameProviderTwice() {
     HashMap<Byte, SaslClientAuthenticationProvider> registeredProviders = new HashMap<>();
@@ -58,38 +53,19 @@ public void testCannotAddTheSameProviderTwice() {
     SaslClientAuthenticationProviders.addProviderIfNotExists(p1, registeredProviders);
     assertEquals(1, registeredProviders.size());
 
-    try {
-      SaslClientAuthenticationProviders.addProviderIfNotExists(p2, registeredProviders);
-    } catch (RuntimeException e) {
-    }
+    assertThrows(RuntimeException.class,
+      () -> SaslClientAuthenticationProviders.addProviderIfNotExists(p2, registeredProviders));
 
-    assertSame("Expected the original provider to be present", p1,
-      registeredProviders.entrySet().iterator().next().getValue());
+    assertSame(p1, registeredProviders.entrySet().iterator().next().getValue(),
+      "Expected the original provider to be present");
   }
 
   @Test
-  public void testInstanceIsCached() {
-    Configuration conf = HBaseConfiguration.create();
-    SaslClientAuthenticationProviders providers1 =
-      SaslClientAuthenticationProviders.getInstance(conf);
-    SaslClientAuthenticationProviders providers2 =
-      SaslClientAuthenticationProviders.getInstance(conf);
-    assertSame(providers1, providers2);
-
-    SaslClientAuthenticationProviders.reset();
-
-    SaslClientAuthenticationProviders providers3 =
-      SaslClientAuthenticationProviders.getInstance(conf);
-    assertNotSame(providers1, providers3);
-    assertEquals(providers1.getNumRegisteredProviders(), providers3.getNumRegisteredProviders());
-  }
-
-  @Test(expected = RuntimeException.class)
   public void testDifferentConflictingImplementationsFail() {
     Configuration conf = HBaseConfiguration.create();
     conf.setStrings(SaslClientAuthenticationProviders.EXTRA_PROVIDERS_KEY,
       ConflictingProvider1.class.getName(), ConflictingProvider2.class.getName());
-    SaslClientAuthenticationProviders.getInstance(conf);
+    assertThrows(RuntimeException.class, () -> new SaslClientAuthenticationProviders(conf));
   }
 
   static class ConflictingProvider1 implements SaslClientAuthenticationProvider {

hbase-common/src/test/java/org/apache/hadoop/hbase/HBaseParameterizedParameterResolver.java

@@ -23,6 +23,8 @@
 import org.junit.jupiter.api.extension.ParameterResolver;
 import org.junit.jupiter.params.provider.Arguments;
 
+import org.apache.hbase.thirdparty.com.google.common.primitives.Primitives;
+
 /**
  * @see HBaseParameterizedTestTemplate
  */
@@ -38,8 +40,20 @@ public class HBaseParameterizedParameterResolver implements ParameterResolver {
   public boolean supportsParameter(ParameterContext pc, ExtensionContext ec)
     throws ParameterResolutionException {
     int index = pc.getIndex();
-    return index < values.length
-      && pc.getParameter().getType().isAssignableFrom(values[index].getClass());
+    if (index >= values.length) {
+      return false;
+    }
+    Object value = values[index];
+    Class<?> expectedType = pc.getParameter().getType();
+    if (expectedType.isPrimitive()) {
+      // primitive type can not accept null value
+      if (value == null) {
+        return false;
+      }
+      // test with wrapper type, otherwise it will always return false
+      return Primitives.wrap(expectedType).isAssignableFrom(value.getClass());
+    }
+    return expectedType.isAssignableFrom(value.getClass());
   }
 
   @Override

hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestTableMapReduceUtil.java

@@ -29,7 +29,6 @@
 import org.apache.hadoop.hbase.HBaseClassTestRule;
 import org.apache.hadoop.hbase.HBaseTestingUtility;
 import org.apache.hadoop.hbase.client.Scan;
-import org.apache.hadoop.hbase.security.provider.SaslClientAuthenticationProviders;
 import org.apache.hadoop.hbase.security.token.AuthenticationTokenIdentifier;
 import org.apache.hadoop.hbase.testclassification.MapReduceTests;
 import org.apache.hadoop.hbase.testclassification.MediumTests;
@@ -43,7 +42,6 @@
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.security.token.TokenIdentifier;
-import org.junit.After;
 import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
@@ -59,11 +57,6 @@ public class TestTableMapReduceUtil {
   public static final HBaseClassTestRule CLASS_RULE =
     HBaseClassTestRule.forClass(TestTableMapReduceUtil.class);
 
-  @After
-  public void after() {
-    SaslClientAuthenticationProviders.reset();
-  }
-
   /*
    * initTableSnapshotMapperJob is tested in {@link TestTableSnapshotInputFormat} because the method
    * depends on an online cluster.

hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcServer.java

@@ -53,6 +53,7 @@
 import org.apache.hadoop.hbase.security.SaslUtil.QualityOfProtection;
 import org.apache.hadoop.hbase.security.User;
 import org.apache.hadoop.hbase.security.UserProvider;
+import org.apache.hadoop.hbase.security.provider.SaslServerAuthenticationProviders;
 import org.apache.hadoop.hbase.security.token.AuthenticationTokenSecretManager;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.apache.hadoop.hbase.util.GsonUtil;
@@ -97,6 +98,7 @@ public abstract class RpcServer implements RpcServerInterface, ConfigurationObse
   private final boolean authorize;
   private volatile boolean isOnlineLogProviderEnabled;
   protected boolean isSecurityEnabled;
+  protected final SaslServerAuthenticationProviders saslProviders;
 
   public static final byte CURRENT_VERSION = 0;
 
@@ -309,6 +311,7 @@ public RpcServer(final Server server, final String name,
       saslProps = Collections.emptyMap();
       serverPrincipal = HConstants.EMPTY_STRING;
     }
+    this.saslProviders = new SaslServerAuthenticationProviders(conf);
 
     this.isOnlineLogProviderEnabled = getIsOnlineLogProviderEnabled(conf);
     this.scheduler = scheduler;