Fixed bug where alters that rely on state are retried but shouldn't (#356)

* Fixed bug where alters that rely on state are retried but shouldn't

Author: patduin

CHANGELOG.md

@@ -1,4 +1,8 @@
 
+## [4.1.3] - 2025-11-26
+### Fixed
+- Fixed bug where alter statements are retried resulting incorrect behaviour due to them not being idempotent.
+
 ## [4.1.2] - 2025-11-03
 ### Changed 
 - `hive-exec` dependency to use shaded jar to avoid Kryo conflicts.

waggle-dance-core/pom.xml

@@ -202,10 +202,22 @@
     <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-common</artifactId>
+      <exclusions>
+        <exclusion>
+          <groupId>org.slf4j</groupId>
+          <artifactId>slf4j-reload4j</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
     <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-mapreduce-client-core</artifactId>
+      <exclusions>
+        <exclusion>
+          <groupId>org.slf4j</groupId>
+          <artifactId>slf4j-reload4j</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
 
     <dependency>

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

@@ -33,7 +33,6 @@
 import com.hotels.bdp.waggledance.client.compatibility.HiveCompatibleThriftHiveMetastoreIfaceFactory;
 import com.hotels.hcommon.hive.metastore.exception.MetastoreUnavailableException;
 
-
 public class DefaultMetaStoreClientFactory implements MetaStoreClientFactory {
 
   static final Class<?>[] INTERFACES = new Class<?>[] { CloseableThriftHiveMetastoreIface.class };
@@ -48,9 +47,9 @@ private static class ReconnectingMetastoreClientInvocationHandler implements Inv
     private HiveUgiArgs cachedUgi = null;
 
     private ReconnectingMetastoreClientInvocationHandler(
-            String name,
-            int maxRetries,
-            AbstractThriftMetastoreClientManager base) {
+        String name,
+        int maxRetries,
+        AbstractThriftMetastoreClientManager base) {
       this.name = name;
       this.maxRetries = maxRetries;
       this.base = base;
@@ -81,8 +80,8 @@ public Object invoke(Object proxy, Method method, Object[] args) throws Throwabl
           cachedUgi = new HiveUgiArgs(user, groups);
           if (base.isOpen()) {
             log
-                    .info("calling #set_ugi (on already open client) for user '{}',  on metastore {}", cachedUgi.getUser(),
-                            name);
+                .info("calling #set_ugi (on already open client) for user '{}',  on metastore {}", cachedUgi.getUser(),
+                    name);
             return doRealCall(method, args, attempt);
           } else {
             // delay call until we get the next non set_ugi call, this helps doing unnecessary calls to Federated
@@ -107,21 +106,25 @@ private Object doRealCall(Method method, Object[] args, int attempt) throws Ille
               base.reconnect(cachedUgi);
               continue;
             }
-            throw new MetastoreUnavailableException("Client " + name + " is not available", realException);
+            log.debug("Client " + name + " is not available");
+            throw realException;
           }
           throw realException;
         }
       } while (++attempt <= maxRetries);
       throw new RuntimeException("Unreachable code");
     }
 
+    /**
+     * Decides whether a method should be retried. Only 'get' methods are retried. Alters/creates methods are not
+     * retried as there are cases where this is not idempotent. TODO Potentially in the future we should remove the
+     * whole retry mechanic and just leave that up to the caller/client.
+     */
     private boolean shouldRetry(Method method) {
-      switch (method.getName()) {
-        case "shutdown":
-          return false;
-        default:
-          return true;
+      if (method.getName().startsWith("get")) {
+        return true;
       }
+      return false;
     }
 
     private void reconnectIfDisconnected() {
@@ -143,10 +146,10 @@ private void reconnectIfDisconnected() {
    */
   @Override
   public CloseableThriftHiveMetastoreIface newInstance(
-          HiveConf hiveConf,
-          String name,
-          int reconnectionRetries,
-          int connectionTimeout) {
+      HiveConf hiveConf,
+      String name,
+      int reconnectionRetries,
+      int connectionTimeout) {
     boolean useSasl = hiveConf.getBoolVar(ConfVars.METASTORE_USE_THRIFT_SASL);
     HiveCompatibleThriftHiveMetastoreIfaceFactory factory = new HiveCompatibleThriftHiveMetastoreIfaceFactory();
     AbstractThriftMetastoreClientManager base = null;
@@ -160,13 +163,13 @@ public CloseableThriftHiveMetastoreIface newInstance(
 
   @VisibleForTesting
   CloseableThriftHiveMetastoreIface newInstance(
-          String name,
-          int reconnectionRetries,  
-          AbstractThriftMetastoreClientManager base) {
+      String name,
+      int reconnectionRetries,
+      AbstractThriftMetastoreClientManager base) {
     ReconnectingMetastoreClientInvocationHandler reconnectingHandler = new ReconnectingMetastoreClientInvocationHandler(
         name, reconnectionRetries, base);
-    return (CloseableThriftHiveMetastoreIface) Proxy.newProxyInstance(getClass().getClassLoader(),
-        INTERFACES, reconnectingHandler);
+    return (CloseableThriftHiveMetastoreIface) Proxy
+        .newProxyInstance(getClass().getClassLoader(), INTERFACES, reconnectingHandler);
   }
 
-}
+}

waggle-dance-core/src/test/java/com/hotels/bdp/waggledance/client/DefaultMetaStoreClientFactoryTest.java

@@ -15,17 +15,20 @@
  */
 package com.hotels.bdp.waggledance.client;
 
+import static com.hotels.bdp.waggledance.client.HiveUgiArgsStub.TEST_ARGS;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.fail;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
-import static com.hotels.bdp.waggledance.client.HiveUgiArgsStub.TEST_ARGS;
-
 import java.util.List;
 
+import org.apache.hadoop.hive.metastore.api.EnvironmentContext;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.api.Table;
 import org.apache.hadoop.hive.metastore.api.ThriftHiveMetastore.Iface;
 import org.apache.thrift.TException;
 import org.apache.thrift.transport.TTransportException;
@@ -36,8 +39,6 @@
 
 import com.google.common.collect.Lists;
 
-import com.hotels.hcommon.hive.metastore.exception.MetastoreUnavailableException;
-
 @RunWith(MockitoJUnitRunner.class)
 public class DefaultMetaStoreClientFactoryTest {
 
@@ -103,6 +104,27 @@ public void defaultMethodCallThrowsTransportExceptionRetries() throws TException
     verify(base).reconnect(null);
   }
 
+  @Test
+  public void manipulatingDataMethodCallThrowsTransportExceptionNoRetries() throws TException {
+    when(base.getClient()).thenReturn(client);
+    EnvironmentContext envContext = new EnvironmentContext();
+    Table table = new Table();
+    String dbName = "db";
+    String tableName = "table";
+    doThrow(new TTransportException("No connection"), new MetaException("Server Error"))
+        .when(client)
+        .alter_table_with_environment_context(dbName, tableName, table, envContext);
+
+    CloseableThriftHiveMetastoreIface iface = factory.newInstance("name", RECONNECTION_RETRIES, base);
+
+    try {
+      iface.alter_table_with_environment_context(dbName, tableName, table, envContext);
+      fail("Expected TTransportException");
+    } catch (TTransportException e) {
+      assertThat(e.getMessage(), is("No connection"));
+    }
+  }
+
   @Test
   public void set_ugi_before_call() throws Exception {
     when(base.getClient()).thenReturn(client);
@@ -141,7 +163,7 @@ public void set_ugi_CalledWhenOpen() throws Exception {
     assertThat(setUgiResult, is(Lists.newArrayList("users!")));
   }
 
-  @Test(expected = MetastoreUnavailableException.class)
+  @Test(expected = TTransportException.class)
   public void shutdownThrowsTransportExceptionNoRetry() throws TException {
     when(base.getClient()).thenReturn(client);
     doThrow(new TTransportException()).when(client).shutdown();
@@ -151,7 +173,7 @@ public void shutdownThrowsTransportExceptionNoRetry() throws TException {
     iface.shutdown();
   }
 
-  @Test(expected = MetastoreUnavailableException.class)
+  @Test(expected = TTransportException.class)
   public void defaultMethodCallThrowsTransportExceptionNoRetriesLeft() throws TException {
     when(base.getClient()).thenReturn(client);
     when(client.getName()).thenThrow(new TTransportException());