API: Close encryption manager after FileIO close failure

api/src/main/java/org/apache/iceberg/encryption/EncryptingFileIO.java

@@ -159,14 +159,10 @@ public Map<String, String> properties() {
 
   @Override
   public void close() {
-    io.close();
-
-    if (em instanceof Closeable) {
-      try {
-        ((Closeable) em).close();
-      } catch (IOException e) {
-        throw new UncheckedIOException("Failed to close encryption manager", e);
-      }
+    try (Closeable closeableManager = em instanceof Closeable ? (Closeable) em : null) {
+      io.close();
+    } catch (IOException e) {
+      throw new UncheckedIOException("Failed to close encryption manager", e);
     }
   }
 

api/src/test/java/org/apache/iceberg/encryption/TestEncryptingFileIO.java

@@ -19,12 +19,15 @@
 package org.apache.iceberg.encryption;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 import static org.mockito.Mockito.withSettings;
 
 import java.io.Closeable;
+import java.io.IOException;
 import java.util.List;
 import java.util.Map;
 import org.apache.iceberg.io.DelegateFileIO;
@@ -151,6 +154,39 @@ public void closeClosesEncryptionManagerWhenCloseable() throws Exception {
     verify((Closeable) em).close();
   }
 
+  @Test
+  public void closeClosesEncryptionManagerWhenFileIOCloseFails() throws Exception {
+    RuntimeException fileIOFailure = new RuntimeException("FileIO close failure");
+    EncryptionManager em =
+        mock(EncryptionManager.class, withSettings().extraInterfaces(Closeable.class));
+    DelegateFileIO delegate = mock(DelegateFileIO.class);
+    doThrow(fileIOFailure).when(delegate).close();
+
+    EncryptingFileIO fileIO = EncryptingFileIO.combine(delegate, em);
+
+    assertThatThrownBy(fileIO::close).isSameAs(fileIOFailure);
+    verify((Closeable) em).close();
+  }
+
+  @Test
+  public void closeSuppressesEncryptionManagerFailureWhenFileIOCloseFails() throws Exception {
+    RuntimeException fileIOFailure = new RuntimeException("FileIO close failure");
+    IOException managerFailure = new IOException("Encryption manager close failure");
+    EncryptionManager em =
+        mock(EncryptionManager.class, withSettings().extraInterfaces(Closeable.class));
+    Closeable closeableManager = (Closeable) em;
+    DelegateFileIO delegate = mock(DelegateFileIO.class);
+    doThrow(fileIOFailure).when(delegate).close();
+    doThrow(managerFailure).when(closeableManager).close();
+
+    EncryptingFileIO fileIO = EncryptingFileIO.combine(delegate, em);
+
+    assertThatThrownBy(fileIO::close)
+        .isSameAs(fileIOFailure)
+        .satisfies(failure -> assertThat(failure.getSuppressed()).containsExactly(managerFailure));
+    verify(closeableManager).close();
+  }
+
   @Test
   public void properties() {
     EncryptionManager em = mock(EncryptionManager.class);