This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 5bcdd3fb3c [common] Resolve the atomic rename under the Kerberos
FileSystem wrapper (#9652)
5bcdd3fb3c is described below
commit 5bcdd3fb3cb140384dde1dad9a971e69844df060
Author: YangJie <[email protected]>
AuthorDate: Fri Sep 11 03:08:50 2026 -0400
[common] Resolve the atomic rename under the Kerberos FileSystem wrapper
(#9652)
---
.../org/apache/paimon/fs/hadoop/HadoopFileIO.java | 54 ++++++++-
.../paimon/fs/hadoop/HadoopSecuredFileSystem.java | 15 +++
.../fs/hadoop/HadoopSecuredFileSystemTest.java | 127 +++++++++++++++++++++
3 files changed, 191 insertions(+), 5 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 3ff241d6c8..3114baad2e 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 class HadoopFileIO implements FileIO,
HadoopOptionsProvider {
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 class HadoopFileIO implements FileIO,
HadoopOptionsProvider {
// 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,15 +444,29 @@ public class HadoopFileIO implements FileIO,
HadoopOptionsProvider {
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(
+ () -> {
+ invokeRename(
+ renameMethod,
+ renameTarget,
+ hadoopTemp,
+ hadoopDst,
+ renameOptions);
+ return null;
+ });
+ } else {
+ invokeRename(renameMethod, renameTarget, hadoopTemp,
hadoopDst, renameOptions);
+ }
renameDone = true;
// TODO: this is a workaround of HADOOP-16255 - remove this when
HADOOP-16255 is
// resolved
tryRemoveCrcFile(hadoopTemp);
return true;
- } catch (InvocationTargetException | IllegalAccessException e) {
- throw new IOException(e);
} finally {
if (!renameDone) {
deleteQuietly(tempPath);
@@ -451,6 +474,27 @@ public class HadoopFileIO implements FileIO,
HadoopOptionsProvider {
}
}
+ /**
+ * Invokes the reflective 3-arg rename and translates reflective failures
to {@link
+ * IOException}. Inside {@link HadoopSecuredFileSystem}'s {@code doAs}, an
escaping {@link
+ * InvocationTargetException} is rewrapped as {@code
UndeclaredThrowableException} and would
+ * surface as a {@link RuntimeException}, bypassing the {@code
IOException} retry in {@code
+ * HintFileUtils.commitHint}.
+ */
+ private static void invokeRename(
+ Method renameMethod,
+ FileSystem renameTarget,
+ org.apache.hadoop.fs.Path src,
+ org.apache.hadoop.fs.Path dst,
+ Options.Rename[] renameOptions)
+ throws IOException {
+ try {
+ renameMethod.invoke(renameTarget, src, dst, renameOptions);
+ } catch (InvocationTargetException | IllegalAccessException e) {
+ throw new IOException(e);
+ }
+ }
+
/** @throws IOException if a fatal exception occurs. Will try to ignore
most exceptions. */
@SuppressWarnings("CatchMayIgnoreException")
private void tryRemoveCrcFile(org.apache.hadoop.fs.Path path) throws
IOException {
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 cbfca1b6d9..45bf32ce87 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 @@ public class HadoopSecuredFileSystem extends FileSystem {
}
}
+ /**
+ * 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 {
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 54de46dc7c..12a267501b 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.catalog.CatalogContext;
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,127 @@ public class HadoopSecuredFileSystemTest {
.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");
+ }
+
+ @Test
+ public void testAtomicRenameSurfacesDelegateIOExceptionThroughDoAs()
throws Exception {
+ File dir = new File(tmp.toFile(), "atomic-failure");
+ assertThat(dir.mkdirs()).isTrue();
+ Path target = new Path(new File(dir, "LATEST").toURI());
+
+ FailingRenameFileSystem delegate = new FailingRenameFileSystem();
+ 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()));
+
+ // Method.invoke wraps the delegate IOException in
InvocationTargetException, which doAs
+ // would rewrap as UndeclaredThrowableException; without translation
this surfaces as a
+ // RuntimeException and escapes HintFileUtils.commitHint's IOException
retry loop.
+ assertThatThrownBy(() -> fileIO.tryAtomicOverwriteViaRename(target,
"content"))
+ .isInstanceOf(IOException.class);
+ }
+
+ /** 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");
+ }
+ }
+ }
+
+ /** A local file system whose public 3-arg rename always fails with {@link
IOException}. */
+ private static class FailingRenameFileSystem extends RawLocalFileSystem {
+
+ @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 {
+ 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));
