make load hive table more robust

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

@@ -58,6 +58,7 @@
 import org.apache.iceberg.hadoop.ConfigProperties;
 import org.apache.iceberg.hadoop.HadoopFileIO;
 import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.thrift.TException;
@@ -155,10 +156,7 @@ protected void doCommit(TableMetadata base, TableMetadata metadata) {
       Table tbl = loadHmsTable();
 
       if (tbl != null) {
-        // If we try to create the table but the metadata location is already set, then we had a concurrent commit
-        if (base == null && tbl.getParameters().get(BaseMetastoreTableOperations.METADATA_LOCATION_PROP) != null) {
-          throw new AlreadyExistsException("Table already exists: %s.%s", database, tableName);
-        }
+        checkConcurrentCommit(base, tbl);
 
         updateHiveTable = true;
         LOG.debug("Committing existing table: {}", fullName);
@@ -190,8 +188,16 @@ protected void doCommit(TableMetadata base, TableMetadata metadata) {
             "exist, this probably happened when using embedded metastore or doesn't create a " +
             "transactional meta table. To fix this, use an alternative metastore", e);
       }
-
-      throw new RuntimeException(String.format("Metastore operation failed for %s.%s", database, tableName), e);
+      if (checkPersisTableSuccess(newMetadataLocation, e, updateHiveTable)) {
+        // Check whether the committing metadatalocation  is same with hive.
+        // Ignore exception  If it's  the same
+        LOG.warn(
+                "Metastore operation is fail for {} . but the commit is done successfully, " +
+                        "please check the log of the hive metastore", e.getMessage(), e);
+        threw = false;
+      } else {
+        throw new RuntimeException(String.format("Metastore operation failed for %s.%s", database, tableName), e);
+      }
 
     } catch (InterruptedException e) {
       Thread.currentThread().interrupt();
@@ -206,7 +212,33 @@ protected void doCommit(TableMetadata base, TableMetadata metadata) {
     }
   }
 
-  private void persistTable(Table hmsTable, boolean updateHiveTable) throws TException, InterruptedException {
+  private void checkConcurrentCommit(TableMetadata base, Table tbl) {
+    // If we try to create the table but the metadata location is already set, then we had a concurrent commit
+    if (base == null && tbl.getParameters().get(BaseMetastoreTableOperations.METADATA_LOCATION_PROP) != null) {
+      throw new AlreadyExistsException("Table already exists: %s.%s", database, tableName);
+    }
+  }
+
+  private boolean checkPersisTableSuccess(String newMetadataLocation, Exception exception, boolean updateHiveTable) {
+    boolean success = false;
+    try {
+      Table table = loadHmsTable();
+      if (table != null && exception instanceof TException && updateHiveTable) {
+        Map<String, String> parameters = table.getParameters();
+        String hiveMetadataLocation = parameters.get(METADATA_LOCATION_PROP);
+        success = checkLocationSameWithHive(newMetadataLocation, hiveMetadataLocation);
+      }
+    } catch (Exception e) {
+      LOG.warn("Checking the newMetadataLocation with hive failed", e);
+    }
+    return success;
+  }
+  @VisibleForTesting
+  boolean checkLocationSameWithHive(String newMetadataLocation, String hiveMetadataLocation) {
+    return newMetadataLocation.equals(hiveMetadataLocation);
+  }
+  @VisibleForTesting
+  void persistTable(Table hmsTable, boolean updateHiveTable) throws TException, InterruptedException {
     if (updateHiveTable) {
       metaClients.run(client -> {
         EnvironmentContext envContext = new EnvironmentContext(

hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCommits.java

@@ -28,9 +28,11 @@
 import org.junit.Test;
 import org.mockito.ArgumentCaptor;
 
+import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.spy;
 
+
 public class TestHiveCommits extends HiveTableBaseTest {
 
   @Test
@@ -41,8 +43,8 @@ public void testSuppressUnlockExceptions() throws TException, InterruptedExcepti
     TableMetadata metadataV1 = ops.current();
 
     table.updateSchema()
-        .addColumn("n", Types.IntegerType.get())
-        .commit();
+            .addColumn("n", Types.IntegerType.get())
+            .commit();
 
     ops.refresh();
 
@@ -66,4 +68,52 @@ public void testSuppressUnlockExceptions() throws TException, InterruptedExcepti
     // the commit must succeed
     Assert.assertEquals(1, ops.current().schema().columns().size());
   }
+
+  @Test
+  public void testIgnoreCommitExceptions() throws TException, InterruptedException {
+    Table table = catalog.loadTable(TABLE_IDENTIFIER);
+    HiveTableOperations ops = (HiveTableOperations) ((HasTableOperations) table).operations();
+
+    TableMetadata metadataV1 = ops.current();
+
+    table.updateSchema()
+            .addColumn("n", Types.IntegerType.get())
+            .commit();
+
+    ops.refresh();
+
+    TableMetadata metadataV2 = ops.current();
+
+    Assert.assertEquals(2, ops.current().schema().columns().size());
+
+    HiveTableOperations spyOps = spy(ops);
+
+    ArgumentCaptor<Long> lockId = ArgumentCaptor.forClass(Long.class);
+    doThrow(new RuntimeException()).when(spyOps).doUnlock(lockId.capture());
+
+    ArgumentCaptor<org.apache.hadoop.hive.metastore.api.Table> tableArgumentCaptor =
+            ArgumentCaptor.forClass(org.apache.hadoop.hive.metastore.api.Table.class);
+    ArgumentCaptor<Boolean> booleanArgumentCaptor = ArgumentCaptor.forClass(Boolean.class);
+    // throw  TException when persistTable
+    doThrow(new TException("test")).when(spyOps).persistTable(tableArgumentCaptor.capture(),
+            booleanArgumentCaptor.capture());
+
+    ArgumentCaptor<String> pendingLocationArgumentCaptor =
+            ArgumentCaptor.forClass(String.class);
+    ArgumentCaptor<String> hiveLocationArgumentCaptor = ArgumentCaptor.forClass(String.class);
+    // but the location is same with hive
+    doReturn(true).when(spyOps).checkLocationSameWithHive(pendingLocationArgumentCaptor.capture(),
+            hiveLocationArgumentCaptor.capture());
+
+    try {
+      spyOps.commit(metadataV2, metadataV1);
+      org.apache.hadoop.hive.metastore.api.Table hiveTable = tableArgumentCaptor.getValue();
+      ops.persistTable(hiveTable, true);
+    } finally {
+      ops.doUnlock(lockId.getValue());
+    }
+    ops.refresh();
+    // the commit must succeed
+    Assert.assertEquals(1, ops.current().schema().columns().size());
+  }
 }