[common] Resolve the atomic rename under the Kerberos FileSystem wrapper

paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java

@@ -400,6 +400,15 @@ public boolean tryAtomicOverwriteViaRename(Path dst, String content) throws IOEx
         org.apache.hadoop.fs.Path hadoopDst = path(dst);
         FileSystem fs = getFileSystem(hadoopDst);
 
+        // HadoopSecuredFileSystem cannot override FileSystem's protected 3-arg rename, so
+        // reflection has to find it on the file system underneath the wrapper.
+        final FileSystem renameTarget;
+        if (fs instanceof HadoopSecuredFileSystem) {
+            renameTarget = ((HadoopSecuredFileSystem) fs).unwrap();
+        } else {
+            renameTarget = fs;
+        }
+
         if (renameMethodRef == null) {
             synchronized (this) {
                 if (renameMethodRef == null) {
@@ -409,7 +418,7 @@ public boolean tryAtomicOverwriteViaRename(Path dst, String content) throws IOEx
                     // DistributedFileSystem and ViewFileSystem override the rename method to public
                     // and implement correct renaming
                     try {
-                        method = ReflectionUtils.getMethod(fs.getClass(), "rename", 3);
+                        method = ReflectionUtils.getMethod(renameTarget.getClass(), "rename", 3);
                     } catch (NoSuchMethodException e) {
                         method = null;
                     }
@@ -435,8 +444,20 @@ public boolean tryAtomicOverwriteViaRename(Path dst, String content) throws IOEx
                 writer.flush();
             }
 
-            renameMethod.invoke(
-                    fs, hadoopTemp, hadoopDst, new Options.Rename[] {Options.Rename.OVERWRITE});
+            Options.Rename[] renameOptions = new Options.Rename[] {Options.Rename.OVERWRITE};
+            if (fs instanceof HadoopSecuredFileSystem) {
+                // the call has to stay inside the wrapper's doAs, or the rename runs as
+                // whoever the current thread is rather than the login user
+                ((HadoopSecuredFileSystem) fs)
+                        .callAsLoginUser(
+                                () -> {
+                                    renameMethod.invoke(
+                                            renameTarget, hadoopTemp, hadoopDst, renameOptions);
+                                    return null;
+                                });
+            } else {
+                renameMethod.invoke(renameTarget, hadoopTemp, hadoopDst, renameOptions);
+            }
             renameDone = true;
             // TODO: this is a workaround of HADOOP-16255 - remove this when HADOOP-16255 is
             // resolved

paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopSecuredFileSystem.java

@@ -196,6 +196,21 @@ private <T> T runSecuredWithIOException(final Callable<T> securedCallable) throw
         }
     }
 
+    /**
+     * The underlying {@link FileSystem} this secured wrapper delegates to. Callers that reach past
+     * the wrapper for a method it cannot override, such as {@link FileSystem}'s protected
+     * three-argument {@code rename}, have to run the call through {@link #callAsLoginUser} so it
+     * still happens as the login user.
+     */
+    public FileSystem unwrap() {
+        return fileSystem;
+    }
+
+    /** Runs the callable as the login user, like every delegating method here does. */
+    public <T> T callAsLoginUser(Callable<T> callable) throws IOException {
+        return runSecuredWithIOException(callable);
+    }
+
     public static FileSystem trySecureFileSystem(
             FileSystem fileSystem, Options options, Configuration configuration)
             throws IOException {

paimon-common/src/test/java/org/apache/paimon/fs/hadoop/HadoopSecuredFileSystemTest.java

@@ -22,12 +22,18 @@
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.options.Options;
 
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.RawLocalFileSystem;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
 import java.io.File;
+import java.io.IOException;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Test for {@link HadoopSecuredFileSystem}. */
 public class HadoopSecuredFileSystemTest {
@@ -78,6 +84,92 @@ public void testReturnOriginalFileSystemWhenSecurityConfigIsIllegal() throws Exc
                 .isNotInstanceOf(HadoopSecuredFileSystem.class);
     }
 
+    @Test
+    public void testUnwrapAndCallAsLoginUser() throws Exception {
+        // tryAtomicOverwriteViaRename has to reach FileSystem's protected 3-arg rename on the
+        // file system under the wrapper, and run the call as the login user.
+        HadoopSecuredFileSystem secured = securedFileSystem();
+
+        assertThat(secured.unwrap()).isNotInstanceOf(HadoopSecuredFileSystem.class);
+        assertThat(secured.<String>callAsLoginUser(() -> "ran")).isEqualTo("ran");
+        assertThatThrownBy(
+                        () ->
+                                secured.callAsLoginUser(
+                                        () -> {
+                                            throw new IOException("rename failed");
+                                        }))
+                .isInstanceOf(IOException.class)
+                .hasMessage("rename failed");
+    }
+
+    @Test
+    public void testAtomicRenameRunsOnTheDelegateAsTheLoginUser() throws Exception {
+        File dir = new File(tmp.toFile(), "atomic");
+        assertThat(dir.mkdirs()).isTrue();
+        Path target = new Path(new File(dir, "LATEST").toURI());
+
+        AtomicRenameFileSystem delegate = new AtomicRenameFileSystem();
+        delegate.initialize(target.toUri(), new Configuration());
+        HadoopFileIO fileIO = new HadoopFileIO(target);
+        Options options = kerberosOptions();
+        fileIO.configure(CatalogContext.create(options));
+        fileIO.setFileSystem(
+                HadoopSecuredFileSystem.trySecureFileSystem(
+                        delegate, options, new Configuration()));
+
+        // Reflection only sees public methods, and the wrapper cannot override FileSystem's
+        // protected 3-arg rename, so this only works if the lookup goes to the delegate.
+        assertThat(fileIO.tryAtomicOverwriteViaRename(target, "content")).isTrue();
+        assertThat(delegate.atomicRenames).isEqualTo(1);
+        assertThat(delegate.renameUser)
+                .isEqualTo(UserGroupInformation.getLoginUser().getUserName());
+        assertThat(fileIO.readFileUtf8(target)).isEqualTo("content");
+    }
+
+    /** A local file system exposing {@link FileSystem}'s 3-arg rename as public. */
+    private static class AtomicRenameFileSystem extends RawLocalFileSystem {
+
+        private int atomicRenames;
+        private String renameUser;
+
+        @Override
+        public void rename(
+                org.apache.hadoop.fs.Path src,
+                org.apache.hadoop.fs.Path dst,
+                org.apache.hadoop.fs.Options.Rename... options)
+                throws IOException {
+            atomicRenames++;
+            renameUser = UserGroupInformation.getCurrentUser().getUserName();
+            if (!rename(src, dst)) {
+                throw new IOException("rename failed");
+            }
+        }
+    }
+
+    private Options kerberosOptions() throws IOException {
+        File keytabFile = new File(tmp.toFile(), "k.keytab");
+        if (!keytabFile.exists()) {
+            assertThat(keytabFile.createNewFile()).isTrue();
+        }
+        Options options = new Options();
+        options.set("security.kerberos.login.principal", "test-user");
+        options.set("security.kerberos.login.keytab", keytabFile.getAbsolutePath());
+        return options;
+    }
+
+    private HadoopSecuredFileSystem securedFileSystem() throws Exception {
+        File keytabFile = new File(tmp.toFile(), "k.keytab");
+        assertThat(keytabFile.createNewFile()).isTrue();
+        Options options = new Options();
+        options.set("security.kerberos.login.principal", "test-user");
+        options.set("security.kerberos.login.keytab", keytabFile.getAbsolutePath());
+
+        org.apache.hadoop.fs.FileSystem fs =
+                createFileIO(options).getFileSystem(new org.apache.hadoop.fs.Path("file:///tmp/t"));
+        assertThat(fs).isInstanceOf(HadoopSecuredFileSystem.class);
+        return (HadoopSecuredFileSystem) fs;
+    }
+
     private HadoopFileIO createFileIO(Options options) {
         HadoopFileIO fileIO = new HadoopFileIO(new Path("file:///tmp/test"));
         fileIO.configure(CatalogContext.create(options));