HBASE-29144 Client request fails for KERBEROS with RpcConnectionRegistry

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;
@@ -112,6 +113,7 @@ public abstract class AbstractRpcClient<T extends RpcConnection> implements RpcC
   protected final String clusterId;
   protected final SocketAddress localAddr;
   protected final MetricsConnection metrics;
+  protected final SaslClientAuthenticationProviders authenticationProviders;
 
   protected final UserProvider userProvider;
   protected final CellBlockBuilder cellBlockBuilder;
@@ -180,6 +182,7 @@ public AbstractRpcClient(Configuration conf, String clusterId, SocketAddress loc
     this.readTO = conf.getInt(SOCKET_TIMEOUT_READ, DEFAULT_SOCKET_TIMEOUT_READ);
     this.writeTO = conf.getInt(SOCKET_TIMEOUT_WRITE, DEFAULT_SOCKET_TIMEOUT_WRITE);
     this.metrics = metrics;
+    this.authenticationProviders = SaslClientAuthenticationProviders.getInstance(conf);
     this.maxConcurrentCallsPerServer =
       conf.getInt(HConstants.HBASE_CLIENT_PERSERVER_REQUESTS_THRESHOLD,
         HConstants.DEFAULT_HBASE_CLIENT_PERSERVER_REQUESTS_THRESHOLD);

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

@@ -213,7 +213,8 @@ 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.cellBlockBuilder, rpcClient.metrics, rpcClient.authenticationProviders,
+      rpcClient.connectionAttributes);
     this.rpcClient = rpcClient;
     this.connectionHeaderPreamble = getConnectionHeaderPreamble();
     ConnectionHeader header = getConnectionHeader();

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

@@ -107,7 +107,8 @@ 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.cellBlockBuilder, rpcClient.metrics, rpcClient.authenticationProviders,
+      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

@@ -121,6 +121,7 @@ abstract class RpcConnection {
   protected RpcConnection(Configuration conf, HashedWheelTimer timeoutTimer, ConnectionId remoteId,
     String clusterId, boolean isSecurityEnabled, Codec codec, CompressionCodec compressor,
     CellBlockBuilder cellBlockBuilder, MetricsConnection metrics,
+    SaslClientAuthenticationProviders authenticationProviders,
     Map<String, byte[]> connectionAttributes) throws IOException {
     this.timeoutTimer = timeoutTimer;
     this.codec = codec;
@@ -133,22 +134,20 @@ protected RpcConnection(Configuration conf, HashedWheelTimer timeoutTimer, Conne
     this.securityInfo = SecurityInfo.getInfo(remoteId.getServiceName());
     this.useSasl = isSecurityEnabled;
 
-    // Choose the correct Token and AuthenticationProvider for this client to use
-    SaslClientAuthenticationProviders providers =
-      SaslClientAuthenticationProviders.getInstance(conf);
+    // Choose the correct Token for this client to use
     Pair<SaslClientAuthenticationProvider, Token<? extends TokenIdentifier>> pair;
     if (useSasl && securityInfo != null) {
-      pair = providers.selectProvider(clusterId, ticket);
+      pair = authenticationProviders.selectProvider(clusterId, ticket);
       if (pair == null) {
         if (LOG.isTraceEnabled()) {
           LOG.trace("Found no valid authentication method from providers={} with tokens={}",
-            providers.toString(), ticket.getTokens());
+            authenticationProviders.toString(), ticket.getTokens());
         }
         throw new RuntimeException("Found no valid authentication method from options");
       }
     } else if (!useSasl) {
       // Hack, while SIMPLE doesn't go via SASL.
-      pair = providers.getSimpleProvider();
+      pair = authenticationProviders.getSimpleProvider();
     } else {
       throw new RuntimeException("Could not compute valid client authentication provider");
     }

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

@@ -33,8 +33,7 @@ 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)}.
+   * should be called prior to calling {@link #selectProvider(String, User)}.
    */
   void configure(Configuration conf,
     Collection<SaslClientAuthenticationProvider> availableProviders);

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

@@ -23,7 +23,6 @@
 import java.util.HashMap;
 import java.util.Optional;
 import java.util.ServiceLoader;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
@@ -48,9 +47,6 @@ 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 =
-    new AtomicReference<>();
-
   private final Collection<SaslClientAuthenticationProvider> providers;
   private final AuthenticationProviderSelector selector;
 
@@ -67,26 +63,6 @@ public int getNumRegisteredProviders() {
     return providers.size();
   }
 
-  /**
-   * Returns a singleton instance of {@link SaslClientAuthenticationProviders}.
-   */
-  public static synchronized SaslClientAuthenticationProviders getInstance(Configuration conf) {
-    SaslClientAuthenticationProviders providers = providersRef.get();
-    if (providers == null) {
-      providers = instantiate(conf);
-      providersRef.set(providers);
-    }
-
-    return providers;
-  }
-
-  /**
-   * Removes the cached singleton instance of {@link SaslClientAuthenticationProviders}.
-   */
-  public static synchronized void reset() {
-    providersRef.set(null);
-  }
-
   /**
    * Adds the given {@code provider} to the set, only if an equivalent provider does not already
    * exist in the set.
@@ -165,7 +141,7 @@ static void addExplicitProviders(Configuration conf,
    * Instantiates all client authentication providers and returns an instance of
    * {@link SaslClientAuthenticationProviders}.
    */
-  static SaslClientAuthenticationProviders instantiate(Configuration conf) {
+  public static SaslClientAuthenticationProviders getInstance(Configuration conf) {
     ServiceLoader<SaslClientAuthenticationProvider> loader =
       ServiceLoader.load(SaslClientAuthenticationProvider.class,
         SaslClientAuthenticationProviders.class.getClassLoader());

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

@@ -68,20 +68,14 @@ public void testCannotAddTheSameProviderTwice() {
   }
 
   @Test
-  public void testInstanceIsCached() {
+  public void testInstanceIsNotCached() {
     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());
+    assertNotSame(providers1, providers2);
+    assertEquals(providers1.getNumRegisteredProviders(), providers2.getNumRegisteredProviders());
   }
 
   @Test(expected = RuntimeException.class)

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

@@ -30,7 +30,6 @@
 import org.apache.hadoop.hbase.HBaseClassTestRule;
 import org.apache.hadoop.hbase.HBaseTestingUtil;
 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;
@@ -44,7 +43,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;
@@ -60,11 +58,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/test/java/org/apache/hadoop/hbase/client/TestBasicReadWriteWithDifferentConnectionRegistries.java

@@ -62,7 +62,7 @@ public class TestBasicReadWriteWithDifferentConnectionRegistries {
   private static final Logger LOG =
     LoggerFactory.getLogger(TestBasicReadWriteWithDifferentConnectionRegistries.class);
 
-  private static final HBaseTestingUtil UTIL = new HBaseTestingUtil();
+  protected static final HBaseTestingUtil UTIL = new HBaseTestingUtil();
 
   public enum RegistryImpl {
     ZK,
@@ -100,11 +100,15 @@ public static void tearDownAfterClass() throws Exception {
     UTIL.shutdownMiniCluster();
   }
 
+  protected Connection createConnectionFromUri(URI uri) throws Exception {
+    return ConnectionFactory.createConnection(uri);
+  }
+
   @Before
   public void setUp() throws Exception {
     switch (impl) {
       case ZK: {
-        Configuration conf = HBaseConfiguration.create();
+        Configuration conf = HBaseConfiguration.create(UTIL.getConfiguration());
         conf.setClass(HConstants.CLIENT_CONNECTION_REGISTRY_IMPL_CONF_KEY,
           ZKConnectionRegistry.class, ConnectionRegistry.class);
         String quorum = UTIL.getZkCluster().getAddress().toString();
@@ -116,7 +120,7 @@ public void setUp() throws Exception {
         break;
       }
       case RPC: {
-        Configuration conf = HBaseConfiguration.create();
+        Configuration conf = HBaseConfiguration.create(UTIL.getConfiguration());
         conf.setClass(HConstants.CLIENT_CONNECTION_REGISTRY_IMPL_CONF_KEY,
           RpcConnectionRegistry.class, ConnectionRegistry.class);
         String bootstrapServers =
@@ -131,14 +135,14 @@ public void setUp() throws Exception {
         String path = UTIL.getConfiguration().get(HConstants.ZOOKEEPER_ZNODE_PARENT);
         URI connectionUri = new URI("hbase+zk://" + quorum + path);
         LOG.info("connect to cluster through connection url: {}", connectionUri);
-        conn = ConnectionFactory.createConnection(connectionUri);
+        conn = createConnectionFromUri(connectionUri);
         break;
       }
       case RPC_URI: {
         URI connectionUri = new URI("hbase+rpc://"
           + UTIL.getMiniHBaseCluster().getMaster().getServerName().getAddress().toString());
         LOG.info("connect to cluster through connection url: {}", connectionUri);
-        conn = ConnectionFactory.createConnection(connectionUri);
+        conn = createConnectionFromUri(connectionUri);
         break;
       }
       default:

hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestBasicReadWriteWithDifferentConnectionRegistriesSecure.java

@@ -0,0 +1,73 @@
+/*
+ * 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.hbase.client;
+
+import java.io.File;
+import java.net.URI;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.security.HBaseKerberosUtils;
+import org.apache.hadoop.hbase.security.access.SecureTestUtil;
+import org.apache.hadoop.hbase.testclassification.ClientTests;
+import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.apache.hadoop.minikdc.MiniKdc;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.experimental.categories.Category;
+
+@Category({ MediumTests.class, ClientTests.class })
+public class TestBasicReadWriteWithDifferentConnectionRegistriesSecure
+  extends TestBasicReadWriteWithDifferentConnectionRegistries {
+
+  @ClassRule
+  public static final HBaseClassTestRule CLASS_RULE =
+    HBaseClassTestRule.forClass(TestBasicReadWriteWithDifferentConnectionRegistriesSecure.class);
+
+  private static final String SERVER_PRINCIPAL = "hbase/localhost";
+
+  private static File KEYTAB_FILE;
+  private static MiniKdc KDC;
+
+  @Override
+  protected Connection createConnectionFromUri(URI uri) throws Exception {
+    return ConnectionFactory.createConnection(uri, UTIL.getConfiguration());
+  }
+
+  @BeforeClass
+  public static void setUpBeforeClass() throws Exception {
+    KEYTAB_FILE = new File(UTIL.getDataTestDir("keytab").toUri().getPath());
+    KDC = UTIL.setupMiniKdc(KEYTAB_FILE);
+    KDC.createPrincipal(KEYTAB_FILE, SERVER_PRINCIPAL);
+
+    final Configuration conf = UTIL.getConfiguration();
+    SecureTestUtil.enableSecurity(conf);
+    HBaseKerberosUtils.setSecuredConfiguration(conf, SERVER_PRINCIPAL + '@' + KDC.getRealm(), null);
+
+    UTIL.startMiniCluster();
+  }
+
+  @AfterClass
+  public static void tearDownAfterClass() throws Exception {
+    UTIL.shutdownMiniCluster();
+    if (KDC != null) {
+      KDC.stop();
+    }
+    KEYTAB_FILE.delete();
+  }
+}