[AMORO-4290][AMS] Propagate internal Iceberg table commit failures (#4291)

Log internal commit errors and propagate CommitFailedException to REST clients instead of swallowing failures. Add unit tests for failure propagation and metadata cleanup errors.

Author: huangjc7

amoro-ams/src/main/java/org/apache/amoro/server/table/internal/IcebergInternalTableOperations.java

@@ -32,6 +32,8 @@
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.LocationProvider;
 import org.apache.iceberg.io.OutputFile;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Map;
 import java.util.Objects;
@@ -41,6 +43,8 @@
 /** Iceberg table operations {@link TableOperations} */
 public class IcebergInternalTableOperations extends PersistentBase implements TableOperations {
 
+  private static final Logger LOG = LoggerFactory.getLogger(IcebergInternalTableOperations.class);
+
   private final ServerTableIdentifier identifier;
 
   private TableMetadata current;
@@ -104,7 +108,13 @@ public void commit(TableMetadata base, TableMetadata metadata) {
       org.apache.amoro.server.table.TableMetadata updatedMetadata = doCommit();
       checkCommitSuccess(updatedMetadata, newMetadataFileLocation);
     } catch (Exception e) {
-      io.deleteFile(newMetadataFileLocation);
+      LOG.error("Commit internal iceberg table failed, try to delete the staged metadata files", e);
+      try {
+        io.deleteFile(newMetadataFileLocation);
+      } catch (Exception cleanupException) {
+        LOG.warn("Delete staged metadata files failed, ignore it", cleanupException);
+      }
+      throw new CommitFailedException(e);
     } finally {
       this.tableMetadata = null;
     }

amoro-ams/src/test/java/org/apache/amoro/server/table/internal/TestIcebergInternalTableOperations.java

@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.amoro.server.table.internal;
+
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.apache.amoro.ServerTableIdentifier;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+public class TestIcebergInternalTableOperations {
+
+  @Test
+  public void testCommitFailureIsPropagated() {
+    FileIO io = mock(FileIO.class);
+    CommitFailedException failure = new CommitFailedException("concurrent commit");
+    when(io.newOutputFile(anyString())).thenThrow(failure);
+
+    IcebergInternalTableOperations operations = newOperations(io);
+    TableMetadata metadata = operations.current();
+
+    CommitFailedException actual =
+        assertThrows(CommitFailedException.class, () -> operations.commit(metadata, metadata));
+
+    assertSame(failure, actual.getCause());
+    verify(io).deleteFile(anyString());
+  }
+
+  @Test
+  public void testCleanupFailureDoesNotMaskCommitFailure() {
+    FileIO io = mock(FileIO.class);
+    CommitFailedException failure = new CommitFailedException("concurrent commit");
+    when(io.newOutputFile(anyString())).thenThrow(failure);
+    doThrow(new RuntimeException("cleanup failed")).when(io).deleteFile(anyString());
+
+    IcebergInternalTableOperations operations = newOperations(io);
+    TableMetadata metadata = operations.current();
+
+    CommitFailedException actual =
+        assertThrows(CommitFailedException.class, () -> operations.commit(metadata, metadata));
+
+    assertSame(failure, actual.getCause());
+    verify(io).deleteFile(anyString());
+  }
+
+  private IcebergInternalTableOperations newOperations(FileIO io) {
+    Schema schema = new Schema(Types.NestedField.required(1, "id", Types.LongType.get()));
+    TableMetadata metadata =
+        TableMetadata.newTableMetadata(
+            schema,
+            PartitionSpec.unpartitioned(),
+            SortOrder.unsorted(),
+            "file:/tmp/amoro-test-table",
+            Collections.emptyMap());
+    IcebergInternalTableOperations operations =
+        spy(
+            new IcebergInternalTableOperations(
+                mock(ServerTableIdentifier.class),
+                mock(org.apache.amoro.server.table.TableMetadata.class),
+                io));
+    doReturn(metadata).when(operations).current();
+    return operations;
+  }
+}