code refactoring completed

HowToKerberize.md

@@ -70,11 +70,6 @@ federated-meta-stores:
   - .*
 ```
 
-In start shell , add jvm properties maybe useful.
-```
--Djavax.security.auth.useSubjectCredsOnly=false
-```
-
 Connect to Waggle Dance via beeline, change ` hive.metastore.uris` in Hive configuration file `hive-site.xml`:
 ```
 <property>
@@ -87,10 +82,6 @@ Connect to Waggle Dance via beeline, change ` hive.metastore.uris` in Hive confi
 
 Waggle Dance should be started by a privileged user with a fresh keytab.
 
-If Waggle Dance throws a GSS exception, you have problem with the keytab file.
-Try to perform `kdestroy` and `kinit` operations and check the keytab file ownership flags.
-
-If the Metastore throws an exception with code -127, Waggle Dance is probably using the wrong authentication policy.
-Check the values in `hive-conf.xml` and make sure that HIVE_HOME and HIVE_CONF_DIR are defined.
-
-Don't forget to restart hive services!
+Just start the service directly, no kinit operation is required. 
+Because the ticket information is saved in jvm instead of being saved in a local file.
+In this way, it can automatically renew without the need for additional operations to renew local tickets.

waggle-dance-core/src/main/java/com/hotels/bdp/waggledance/client/AbstractThriftMetastoreClientManager.java

@@ -0,0 +1,174 @@
+/**
+ * Copyright (C) 2016-2024 Expedia, Inc.
+ *
+ * Licensed 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 com.hotels.bdp.waggledance.client;
+
+import java.io.Closeable;
+import java.net.URI;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.hadoop.hive.conf.HiveConfUtil;
+import org.apache.hadoop.hive.metastore.api.ThriftHiveMetastore;
+import org.apache.thrift.TException;
+import org.apache.thrift.transport.TTransport;
+
+import lombok.extern.log4j.Log4j2;
+
+import com.hotels.bdp.waggledance.client.compatibility.HiveCompatibleThriftHiveMetastoreIfaceFactory;
+
+@Log4j2
+public abstract class AbstractThriftMetastoreClientManager implements Closeable {
+
+
+  static final AtomicInteger CONN_COUNT = new AtomicInteger(0);
+  final HiveConf conf;
+  final HiveCompatibleThriftHiveMetastoreIfaceFactory hiveCompatibleThriftHiveMetastoreIfaceFactory;
+  final URI[] metastoreUris;
+  ThriftHiveMetastore.Iface client = null;
+  TTransport transport = null;
+  boolean isConnected = false;
+  // for thrift connects
+  int retries = 5;
+  long retryDelaySeconds = 0;
+
+  final int connectionTimeout;
+  final String msUri;
+
+  AbstractThriftMetastoreClientManager(
+      HiveConf conf,
+      HiveCompatibleThriftHiveMetastoreIfaceFactory hiveCompatibleThriftHiveMetastoreIfaceFactory,
+      int connectionTimeout) {
+    this.conf = conf;
+    this.hiveCompatibleThriftHiveMetastoreIfaceFactory = hiveCompatibleThriftHiveMetastoreIfaceFactory;
+    this.connectionTimeout = connectionTimeout;
+    msUri = conf.getVar(ConfVars.METASTOREURIS);
+
+    if (HiveConfUtil.isEmbeddedMetaStore(msUri)) {
+      throw new RuntimeException("You can't waggle an embedded metastore");
+    }
+
+    // get the number retries
+    retries = HiveConf.getIntVar(conf, ConfVars.METASTORETHRIFTCONNECTIONRETRIES);
+    retryDelaySeconds = conf.getTimeVar(ConfVars.METASTORE_CLIENT_CONNECT_RETRY_DELAY, TimeUnit.SECONDS);
+
+    // user wants file store based configuration
+    if (msUri != null) {
+      String[] metastoreUrisString = msUri.split(",");
+      metastoreUris = new URI[metastoreUrisString.length];
+      try {
+        int i = 0;
+        for (String s : metastoreUrisString) {
+          URI tmpUri = new URI(s);
+          if (tmpUri.getScheme() == null) {
+            throw new IllegalArgumentException("URI: " + s + " does not have a scheme");
+          }
+          metastoreUris[i++] = tmpUri;
+        }
+      } catch (IllegalArgumentException e) {
+        throw (e);
+      } catch (Exception e) {
+        String exInfo = "Got exception: " + e.getClass().getName() + " " + e.getMessage();
+        log.error(exInfo, e);
+        throw new RuntimeException(exInfo, e);
+      }
+    } else {
+      log.error("NOT getting uris from conf");
+      throw new RuntimeException("MetaStoreURIs not found in conf file");
+    }
+  }
+
+  void open() {
+    open(null);
+  }
+
+  abstract void open(HiveUgiArgs ugiArgs);
+
+  void reconnect(HiveUgiArgs ugiArgs) {
+    close();
+    // Swap the first element of the metastoreUris[] with a random element from the rest
+    // of the array. Rationale being that this method will generally be called when the default
+    // connection has died and the default connection is likely to be the first array element.
+    promoteRandomMetaStoreURI();
+    open(ugiArgs);
+  }
+
+  public String getHiveConfValue(String key, String defaultValue) {
+    return conf.get(key, defaultValue);
+  }
+
+  public void setHiveConfValue(String key, String value) {
+    conf.set(key, value);
+  }
+
+  public String generateNewTokenSignature(String defaultTokenSignature) {
+    String tokenSignature = conf.get(ConfVars.METASTORE_TOKEN_SIGNATURE.varname,
+        defaultTokenSignature);
+    conf.set(ConfVars.METASTORE_TOKEN_SIGNATURE.varname,
+        tokenSignature);
+    return tokenSignature;
+  }
+
+  public Boolean isSaslEnabled() {
+    return conf.getBoolVar(ConfVars.METASTORE_USE_THRIFT_SASL);
+  }
+
+  @Override
+  public void close() {
+    if (!isConnected) {
+      return;
+    }
+    isConnected = false;
+    try {
+      if (client != null) {
+        client.shutdown();
+      }
+    } catch (TException e) {
+      log.debug("Unable to shutdown metastore client. Will try closing transport directly.", e);
+    }
+    // Transport would have got closed via client.shutdown(), so we don't need this, but
+    // just in case, we make this call.
+    if ((transport != null) && transport.isOpen()) {
+      transport.close();
+      transport = null;
+    }
+    log.info("Closed a connection to metastore, current connections: {}", CONN_COUNT.decrementAndGet());
+  }
+
+  boolean isOpen() {
+    return (transport != null) && transport.isOpen();
+  }
+
+  protected ThriftHiveMetastore.Iface getClient() {
+    return client;
+  }
+
+  /**
+   * Swaps the first element of the metastoreUris array with a random element from the remainder of the array.
+   */
+  private void promoteRandomMetaStoreURI() {
+    if (metastoreUris.length <= 1) {
+      return;
+    }
+    Random rng = new Random();
+    int index = rng.nextInt(metastoreUris.length - 1) + 1;
+    URI tmp = metastoreUris[0];
+    metastoreUris[0] = metastoreUris[index];
+    metastoreUris[index] = tmp;
+  }
+}

waggle-dance-core/src/main/java/com/hotels/bdp/waggledance/client/DefaultMetaStoreClientFactory.java

@@ -22,6 +22,7 @@
 import java.util.List;
 
 import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
 import org.apache.thrift.transport.TTransportException;
 
 import lombok.extern.log4j.Log4j2;
@@ -40,7 +41,7 @@ public class DefaultMetaStoreClientFactory implements MetaStoreClientFactory {
   @Log4j2
   private static class ReconnectingMetastoreClientInvocationHandler implements InvocationHandler {
 
-    private final ThriftMetastoreClientManager base;
+    private final AbstractThriftMetastoreClientManager base;
     private final String name;
     private final int maxRetries;
 
@@ -49,7 +50,7 @@ private static class ReconnectingMetastoreClientInvocationHandler implements Inv
     private ReconnectingMetastoreClientInvocationHandler(
             String name,
             int maxRetries,
-            ThriftMetastoreClientManager base) {
+        AbstractThriftMetastoreClientManager base) {
       this.name = name;
       this.maxRetries = maxRetries;
       this.base = base;
@@ -94,7 +95,7 @@ public Object invoke(Object proxy, Method method, Object[] args) throws Throwabl
       }
     }
 
-    private Object doRealCall(Method method, Object[] args, int attempt) throws IllegalAccessException, Throwable {
+    private Object doRealCall(Method method, Object[] args, int attempt) throws Throwable {
       do {
         try {
           return method.invoke(base.getClient(), args);
@@ -146,15 +147,22 @@ public CloseableThriftHiveMetastoreIface newInstance(
           String name,
           int reconnectionRetries,
           int connectionTimeout) {
-    return newInstance(name, reconnectionRetries, new ThriftMetastoreClientManager(hiveConf,
-            new HiveCompatibleThriftHiveMetastoreIfaceFactory(), connectionTimeout));
+    boolean useSasl = hiveConf.getBoolVar(ConfVars.METASTORE_USE_THRIFT_SASL);
+    HiveCompatibleThriftHiveMetastoreIfaceFactory factory = new HiveCompatibleThriftHiveMetastoreIfaceFactory();
+    AbstractThriftMetastoreClientManager base = null;
+    if (useSasl) {
+      base = new SaslThriftMetastoreClientManager(hiveConf, factory, connectionTimeout);
+    } else {
+      base = new ThriftMetastoreClientManager(hiveConf, factory, connectionTimeout);
+    }
+    return newInstance(name, reconnectionRetries, base);
   }
 
   @VisibleForTesting
   CloseableThriftHiveMetastoreIface newInstance(
           String name,
           int reconnectionRetries,
-          ThriftMetastoreClientManager base) {
+          AbstractThriftMetastoreClientManager base) {
     ReconnectingMetastoreClientInvocationHandler reconnectingHandler = new ReconnectingMetastoreClientInvocationHandler(
         name, reconnectionRetries, base);
     return (CloseableThriftHiveMetastoreIface) Proxy.newProxyInstance(getClass().getClassLoader(),