Remove unnecessary Failure check in doCommit path

aws/src/integration/java/org/apache/iceberg/aws/glue/TestGlueCatalogCommitFailure.java

@@ -336,7 +336,7 @@ private void concurrentCommitAndThrowException(
                   mapProperties.get(BaseMetastoreTableOperations.METADATA_LOCATION_PROP);
 
               // Simulate lock expiration or removal, use commit status null to avoid deleting data
-              realOps.cleanupMetadataAndUnlock(null, newMetadataLocation);
+              realOps.unlockMetadataFile(newMetadataLocation);
 
               table.refresh();
               table.updateSchema().addColumn("newCol", Types.IntegerType.get()).commit();

aws/src/main/java/org/apache/iceberg/aws/dynamodb/DynamoDbTableOperations.java

@@ -144,21 +144,9 @@ protected void doCommit(TableMetadata base, TableMetadata metadata) {
       switch (commitStatus) {
         case SUCCESS:
           break;
-        case FAILURE:
-          throw new CommitFailedException(
-              persistFailure, "Cannot commit %s due to unexpected exception", tableName());
         case UNKNOWN:
           throw new CommitStateUnknownException(persistFailure);
       }
-    } finally {
-      try {
-        if (commitStatus == CommitStatus.FAILURE) {
-          // if anything went wrong, clean up the uncommitted metadata file
-          io().deleteFile(newMetadataLocation);
-        }
-      } catch (RuntimeException e) {
-        LOG.error("Failed to cleanup metadata file at {}", newMetadataLocation, e);
-      }
     }
   }
 

aws/src/main/java/org/apache/iceberg/aws/glue/GlueTableOperations.java

@@ -183,14 +183,11 @@ protected void doCommit(TableMetadata base, TableMetadata metadata) {
       switch (commitStatus) {
         case SUCCESS:
           break;
-        case FAILURE:
-          throw new CommitFailedException(
-              persistFailure, "Cannot commit %s due to unexpected exception", tableName());
         case UNKNOWN:
           throw new CommitStateUnknownException(persistFailure);
       }
     } finally {
-      cleanupMetadataAndUnlock(commitStatus, newMetadataLocation);
+      unlockMetadataFile(newMetadataLocation);
       cleanupGlueTempTableIfNecessary(glueTempTableCreated, commitStatus);
     }
   }
@@ -385,20 +382,9 @@ private void handleAWSExceptions(AwsServiceException persistFailure) {
   }
 
   @VisibleForTesting
-  void cleanupMetadataAndUnlock(CommitStatus commitStatus, String metadataLocation) {
-    try {
-      if (commitStatus == CommitStatus.FAILURE
-          && metadataLocation != null
-          && !metadataLocation.isEmpty()) {
-        // if anything went wrong, clean up the uncommitted metadata file
-        io().deleteFile(metadataLocation);
-      }
-    } catch (RuntimeException e) {
-      LOG.error("Failed to cleanup metadata file at {}", metadataLocation, e);
-    } finally {
-      if (lockManager != null) {
-        lockManager.release(commitLockEntityId, metadataLocation);
-      }
+  void unlockMetadataFile(String metadataLocation) {
+    if (lockManager != null) {
+      lockManager.release(commitLockEntityId, metadataLocation);
     }
   }
 

core/src/main/java/org/apache/iceberg/BaseMetastoreTableOperations.java

@@ -306,7 +306,7 @@ protected enum CommitStatus {
    *
    * @param newMetadataLocation the path of the new commit file
    * @param config metadata to use for configuration
-   * @return Commit Status of Success, Failure or Unknown
+   * @return Commit Status of Success or Unknown
    */
   protected CommitStatus checkCommitStatus(String newMetadataLocation, TableMetadata config) {
     int maxAttempts =