Repository: apache/iceberg (language: python)

Issue:
Incorrect Deletion of Snapshot Metadata Due to OutOfMemoryError
### Apache Iceberg version

1.1.0

### Query engine

Flink

### Please describe the bug 🐞

when calling checkCommitStatus method, unexpected errors maybe occur,  such as OutOfMemoryError, during the checkCommitStatus method execution.  the code show as below:
```
commitStatus =
   checkCommitStatus(
       viewName,
       newMetadataLocation,
       metadata.properties(),
       () -> checkCurrentMetadataLocation(newMetadataLocation));
```
During the execution of the `org.apache.iceberg.hive.HiveViewOperations.checkCurrentMetadataLocation` method's refresh operation to download and update the current table metadata, memory consumption occurs, potentially leading to an `OutOfMemoryError`. 
It is important to note that `Tasks` may not handle `Error` exceptions, instead throwing them directly. 
Finally, in the `finally` block, the cleanupMetadataAndUnlock function may delete the table snapshot metadata file, even if it has been recently committed.

### Willingness to contribute

- [X] I can contribute a fix for this bug independently
- [X] I would be willing to contribute a fix for this bug with guidance from the Iceberg community
- [ ] I cannot contribute a fix for this bug at this time


Discussion / hints:

Fix patch:
diff --git a/hive-metastore/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java b/hive-metastore/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java
index 518daaf6acd1..619f20ab87a3 100644
--- a/hive-metastore/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java
+++ b/hive-metastore/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java
@@ -292,6 +292,7 @@ protected void doCommit(TableMetadata base, TableMetadata metadata) {
             database,
             tableName,
             e);
+        commitStatus = BaseMetastoreOperations.CommitStatus.UNKNOWN;
         commitStatus =
             BaseMetastoreOperations.CommitStatus.valueOf(
                 checkCommitStatus(newMetadataLocation, metadata).name());
diff --git a/hive-metastore/src/main/java/org/apache/iceberg/hive/HiveViewOperations.java b/hive-metastore/src/main/java/org/apache/iceberg/hive/HiveViewOperations.java
index 4fc71299d457..dd0d5b3132d0 100644
--- a/hive-metastore/src/main/java/org/apache/iceberg/hive/HiveViewOperations.java
+++ b/hive-metastore/src/main/java/org/apache/iceberg/hive/HiveViewOperations.java
@@ -33,6 +33,7 @@
 import org.apache.hadoop.hive.metastore.api.InvalidObjectException;
 import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
 import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.iceberg.BaseMetastoreOperations;
 import org.apache.iceberg.BaseMetastoreTableOperations;
 import org.apache.iceberg.CatalogUtil;
 import org.apache.iceberg.ClientPool;
@@ -226,6 +227,7 @@ public void doCommit(ViewMetadata base, ViewMetadata metadata) {
             database,
             viewName,
             e);
+        commitStatus = BaseMetastoreOperations.CommitStatus.UNKNOWN;
         commitStatus =
             checkCommitStatus(
                 viewName,


Test patch:
diff --git a/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCommits.java b/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCommits.java
index 136c96934189..754ed55e81e8 100644
--- a/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCommits.java
+++ b/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCommits.java
@@ -21,6 +21,7 @@
 import static org.apache.iceberg.TableProperties.HIVE_LOCK_ENABLED;
 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.anyBoolean;
 import static org.mockito.Mockito.doAnswer;
@@ -44,6 +45,7 @@
 import org.apache.iceberg.types.Types;
 import org.apache.thrift.TException;
 import org.junit.jupiter.api.Test;
+import org.junit.platform.commons.support.ReflectionSupport;
 
 public class TestHiveCommits extends HiveTableBaseTest {
 
@@ -399,6 +401,59 @@ public void testLockExceptionUnknownSuccessCommit() throws TException, Interrupt
         .isTrue();
   }
 
+  @Test
+  public void testSuccessCommitWhenCheckCommitStatusOOM() 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();
+
+    assertThat(ops.current().schema().columns()).hasSize(2);
+
+    HiveTableOperations spyOps = spy(ops);
+
+    // Simulate a communication error after a successful commit
+    doAnswer(
+            i -> {
+              org.apache.hadoop.hive.metastore.api.Table tbl =
+                  i.getArgument(0, org.apache.hadoop.hive.metastore.api.Table.class);
+              String location = i.getArgument(2, String.class);
+              ops.persistTable(tbl, true, location);
+              throw new UnknownError();
+            })
+        .when(spyOps)
+        .persistTable(any(), anyBoolean(), any());
+    try {
+      ReflectionSupport.invokeMethod(
+          ops.getClass()
+              .getSuperclass()
+              .getDeclaredMethod("checkCommitStatus", String.class, TableMetadata.class),
+          doThrow(new OutOfMemoryError()).when(spyOps),
+          anyString(),
+          any());
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+
+    assertThatThrownBy(() -> spyOps.commit(metadataV2, metadataV1))
+        .isInstanceOf(OutOfMemoryError.class);
+
+    ops.refresh();
+
+    assertThat(ops.current().location())
+        .as("Current metadata should have changed to metadata V1")
+        .isEqualTo(metadataV1.location());
+    assertThat(metadataFileExists(ops.current()))
+        .as("Current metadata file should still exist")
+        .isTrue();
+  }
+
   @Test
   public void testCommitExceptionWithoutMessage() throws TException, InterruptedException {
     Table table = catalog.loadTable(TABLE_IDENTIFIER);
diff --git a/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveViewCommits.java b/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveViewCommits.java
index 47abb51602fa..ae251aacebca 100644
--- a/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveViewCommits.java
+++ b/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveViewCommits.java
@@ -21,6 +21,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.anyBoolean;
 import static org.mockito.Mockito.doAnswer;
@@ -31,8 +32,10 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
 import org.apache.hadoop.fs.Path;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.CatalogUtil;
@@ -53,6 +56,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.platform.commons.support.ReflectionSupport;
 
 /** Test Hive locks and Hive errors and retry during commits. */
 public class TestHiveViewCommits {
@@ -434,6 +438,60 @@ public void testLockExceptionUnknownSuccessCommit() throws TException, Interrupt
     assertThat(metadataFileCount(metadataV2)).as("New metadata file should exist").isEqualTo(2);
   }
 
+  @Test
+  public void testSuccessCommitWhenCheckCommitStatusOOM() throws TException, InterruptedException {
+    HiveViewOperations ops = (HiveViewOperations) ((BaseView) view).operations();
+    ViewMetadata metadataV1 = ops.current();
+    assertThat(metadataV1.properties()).hasSize(0);
+
+    view.updateProperties().set("k1", "v1").commit();
+    ops.refresh();
+    ViewMetadata metadataV2 = ops.current();
+    assertThat(metadataV2.properties()).hasSize(1).containsEntry("k1", "v1");
+
+    HiveViewOperations spyOps = spy(ops);
+
+    // Simulate a communication error after a successful commit
+    doAnswer(
+            i -> {
+              org.apache.hadoop.hive.metastore.api.Table tbl =
+                  i.getArgument(0, org.apache.hadoop.hive.metastore.api.Table.class);
+              String location = i.getArgument(2, String.class);
+              ops.persistTable(tbl, true, location);
+              throw new UnknownError();
+            })
+        .when(spyOps)
+        .persistTable(any(), anyBoolean(), any());
+    try {
+      ReflectionSupport.invokeMethod(
+          ops.getClass()
+              .getSuperclass()
+              .getSuperclass()
+              .getDeclaredMethod(
+                  "checkCommitStatus", String.class, String.class, Map.class, Supplier.class),
+          doThrow(new OutOfMemoryError()).when(spyOps),
+          anyString(),
+          anyString(),
+          any(),
+          any());
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+
+    assertThatThrownBy(() -> spyOps.commit(metadataV2, metadataV1))
+        .isInstanceOf(OutOfMemoryError.class);
+
+    ops.refresh();
+
+    assertThat(metadataV2.location())
+        .as("Current metadata should have changed to metadata V1")
+        .isEqualTo(metadataV1.location());
+    assertThat(metadataFileExists(metadataV2))
+        .as("Current metadata file should still exist")
+        .isTrue();
+    assertThat(metadataFileCount(metadataV2)).as("New metadata file should exist").isEqualTo(2);
+  }
+
   @Test
   public void testCommitExceptionWithoutMessage() throws TException, InterruptedException {
     HiveViewOperations ops = (HiveViewOperations) ((BaseView) view).operations();