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));