Core, Hive: Double check commit status in case of commit conflict for NoLock (#12637)

Author: kasakrisz

iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java

@@ -222,14 +222,6 @@ protected void doCommit(TableMetadata base, TableMetadata metadata) {
         throw e;
 
       } catch (Throwable e) {
-        if (e.getMessage() != null && e.getMessage().contains(
-            "The table has been modified. The parameter value for key '" +
-              HiveTableOperations.METADATA_LOCATION_PROP +
-              "' is")) {
-          throw new CommitFailedException(
-              e, "The table %s.%s has been modified concurrently", database, tableName);
-        }
-
         if (e.getMessage() != null && e.getMessage().contains("Table/View 'HIVE_LOCKS' does not exist")) {
           throw new RuntimeException(
               "Failed to acquire locks from metastore because the underlying metastore " +
@@ -238,12 +230,28 @@ protected void doCommit(TableMetadata base, TableMetadata metadata) {
               e);
         }
 
-        LOG.error("Cannot tell if commit to {}.{} succeeded, attempting to reconnect and check.",
-            database, tableName, e);
         commitStatus = BaseMetastoreOperations.CommitStatus.UNKNOWN;
-        commitStatus =
-                BaseMetastoreOperations.CommitStatus.valueOf(
-                        checkCommitStatus(newMetadataLocation, metadata).name());
+        if (e.getMessage() != null && e.getMessage().contains(
+                "The table has been modified. The parameter value for key '" +
+                        HiveTableOperations.METADATA_LOCATION_PROP +
+                        "' is")) {
+          // It's possible the HMS client incorrectly retries a successful operation, due to network
+          // issue for example, and triggers this exception. So we need double-check to make sure
+          // this is really a concurrent modification. Hitting this exception means no pending
+          // requests, if any, can succeed later, so it's safe to check status in strict mode
+          commitStatus = checkCommitStatusStrict(newMetadataLocation, metadata);
+          if (commitStatus == BaseMetastoreOperations.CommitStatus.FAILURE) {
+            throw new CommitFailedException(
+                    e, "The table %s.%s has been modified concurrently", database, tableName);
+          }
+        } else {
+          LOG.error(
+                  "Cannot tell if commit to {}.{} succeeded, attempting to reconnect and check.",
+                  database,
+                  tableName,
+                  e);
+          commitStatus = checkCommitStatus(newMetadataLocation, metadata);
+        }
         switch (commitStatus) {
           case SUCCESS:
             break;

iceberg/iceberg-catalog/src/test/java/org/apache/iceberg/hive/HiveMetastoreExtension.java

@@ -49,7 +49,7 @@ public void beforeAll(ExtensionContext extensionContext) throws Exception {
       }
     }
 
-    metastore.start(hiveConfWithOverrides);
+    metastore.start(hiveConfWithOverrides, 5, true);
     metastoreClient = new HiveMetaStoreClient(hiveConfWithOverrides);
     if (null != databaseName) {
       String dbPath = metastore.getDatabasePath(databaseName);

iceberg/iceberg-catalog/src/test/java/org/apache/iceberg/hive/TestHiveCommitLocks.java

@@ -44,6 +44,7 @@
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.hadoop.ConfigProperties;
@@ -66,6 +67,7 @@
 import static org.apache.iceberg.types.Types.NestedField.required;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.doAnswer;
@@ -206,6 +208,38 @@ public static void cleanup() {
     }
   }
 
+  @Test
+  public void testMultipleAlterTableForNoLock() throws Exception {
+    Table table = catalog.loadTable(TABLE_IDENTIFIER);
+    table.updateProperties().set(TableProperties.HIVE_LOCK_ENABLED, "false").commit();
+    spyOps.refresh();
+    TableMetadata metadataV3 = spyOps.current();
+    AtomicReference<Throwable> alterTableException = new AtomicReference<>(null);
+    doAnswer(
+            i -> {
+              try {
+                // mock a situation where alter table is unexpectedly invoked more than once
+                i.callRealMethod();
+                return i.callRealMethod();
+              } catch (Throwable e) {
+                alterTableException.compareAndSet(null, e);
+                throw e;
+              }
+            })
+            .when(spyClient)
+            .alter_table_with_environmentContext(anyString(), anyString(), any(), any());
+    spyOps.commit(metadataV3, metadataV1);
+    verify(spyClient, times(1))
+            .alter_table_with_environmentContext(anyString(), anyString(), any(), any());
+    assertThat(alterTableException)
+            .as("Expecting to trigger an exception indicating table has been modified")
+            .hasValueMatching(
+                    t ->
+                            t.getMessage()
+                                    .contains("The table has been modified. The parameter value for key '"));
+  }
+
+
   @Test
   public void testLockAcquisitionAtFirstTime() throws TException, InterruptedException {
     doReturn(acquiredLockResponse).when(spyClient).lock(any());

iceberg/iceberg-catalog/src/test/java/org/apache/iceberg/hive/TestHiveMetastore.java

@@ -140,10 +140,21 @@ public void start(HiveConf conf) {
    * @param poolSize The number of threads in the executor pool
    */
   public void start(HiveConf conf, int poolSize) {
+    start(conf, poolSize, false);
+  }
+
+  /**
+   * Starts a TestHiveMetastore with a provided connection pool size and HiveConf.
+   *
+   * @param conf The hive configuration to use
+   * @param poolSize The number of threads in the executor pool
+   * @param directSql Used to turn on directSql
+   */
+  public void start(HiveConf conf, int poolSize, boolean directSql) {
     try {
       TServerSocket socket = new TServerSocket(0);
       int port = socket.getServerSocket().getLocalPort();
-      initConf(conf, port);
+      initConf(conf, port, directSql);
 
       this.hiveConf = conf;
       this.server = newThriftServer(socket, poolSize, hiveConf);
@@ -243,12 +254,12 @@ private TServer newThriftServer(TServerSocket socket, int poolSize, HiveConf con
     return new TThreadPoolServer(args);
   }
 
-  private void initConf(HiveConf conf, int port) {
+  private void initConf(HiveConf conf, int port, boolean directSql) {
     conf.set(HiveConf.ConfVars.METASTORE_URIS.varname, "thrift://localhost:" + port);
     conf.set(HiveConf.ConfVars.METASTORE_WAREHOUSE.varname, "file:" + HIVE_WAREHOUSE_DIR.getAbsolutePath());
     conf.set(HiveConf.ConfVars.HIVE_METASTORE_WAREHOUSE_EXTERNAL.varname,
         "file:" + HIVE_EXTERNAL_WAREHOUSE_DIR.getAbsolutePath());
-    conf.set(HiveConf.ConfVars.METASTORE_TRY_DIRECT_SQL.varname, "false");
+    conf.set(HiveConf.ConfVars.METASTORE_TRY_DIRECT_SQL.varname, String.valueOf(directSql));
     conf.set(HiveConf.ConfVars.METASTORE_DISALLOW_INCOMPATIBLE_COL_TYPE_CHANGES.varname, "false");
     conf.set("iceberg.hive.client-pool-size", "2");
     // set to false so that TxnManager#checkLock does not throw exception when using UNSET data type operation