From 2bcd055a05235ad54829e83bca79c09d20a1b0d8 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Sat, 5 Sep 2026 10:00:05 +0800 Subject: [PATCH] [common] Resolve the atomic rename under the Kerberos FileSystem wrapper tryAtomicOverwriteViaRename looks up FileSystem's 3-arg rename with ReflectionUtils.getMethod, which only sees public methods. HadoopSecuredFileSystem overrides just the 2-arg rename, so with Kerberos configured the lookup failed and the method returned false for good: hint files, consumer resets, tags and _SUCCESS were all written by in-place overwrite instead, silently. Resolve and invoke on the file system under the wrapper, with the invocation inside the wrapper's doAs so the rename still runs as the login user like the temp file it renames. --- .../apache/paimon/fs/hadoop/HadoopFileIO.java | 27 +++++- .../fs/hadoop/HadoopSecuredFileSystem.java | 15 +++ .../hadoop/HadoopSecuredFileSystemTest.java | 92 +++++++++++++++++++ 3 files changed, 131 insertions(+), 3 deletions(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java index 3ff241d6c8f2..537496de3518 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java +++ b/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 diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopSecuredFileSystem.java b/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopSecuredFileSystem.java index cbfca1b6d953..45bf32ce87d6 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopSecuredFileSystem.java +++ b/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopSecuredFileSystem.java @@ -196,6 +196,21 @@ private T runSecuredWithIOException(final Callable 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 callAsLoginUser(Callable callable) throws IOException { + return runSecuredWithIOException(callable); + } + public static FileSystem trySecureFileSystem( FileSystem fileSystem, Options options, Configuration configuration) throws IOException { diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/hadoop/HadoopSecuredFileSystemTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/hadoop/HadoopSecuredFileSystemTest.java index 54de46dc7c56..8b80f0adcab2 100644 --- a/paimon-common/src/test/java/org/apache/paimon/fs/hadoop/HadoopSecuredFileSystemTest.java +++ b/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.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));