Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view

This file was deleted.

This file was deleted.

Original file line number Diff line number Diff line change
Expand Up @@ -19,23 +19,45 @@
package org.apache.gravitino.hive.kerberos;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.file.Files;
import java.nio.file.StandardCopyOption;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class FetchFileUtils {

/**
* Per-destination lock map used to serialize concurrent symlink creation for the same keytab
* file. Keyed by the normalized absolute destination path string to avoid races caused by
* different path spellings referring to the same file. Entries are removed when the corresponding
* {@link KerberosClient} is closed, so the map size is bounded by the number of live catalogs.
*/
private static final ConcurrentHashMap<String, Object> SYMLINK_LOCKS = new ConcurrentHashMap<>();

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When will it be clear?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Close at the Kerberos client closed

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I did not see related close logic on it. As it's a static field, I believe it will only be clear when the classloader is closed.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I added a function removeLock, which is called when the Kerberos client closes.


Comment thread
diqiu50 marked this conversation as resolved.
private FetchFileUtils() {}

/**
* Removes the per-destination lock entry for the given file. Should be called when the keytab
* file is deleted (e.g., on {@link KerberosClient#close()}) to prevent unbounded map growth.
*
* @param destFile the keytab destination file whose lock entry should be removed
*/
static void removeLock(File destFile) {
SYMLINK_LOCKS.remove(destFile.toPath().toAbsolutePath().normalize().toString());
}

public static void fetchFileFromUri(
String fileUri, File destFile, int timeout, Configuration conf) throws java.io.IOException {
String fileUri, File destFile, int timeout, Configuration conf) throws IOException {
try {
URI uri = new URI(fileUri);
String scheme = java.util.Optional.ofNullable(uri.getScheme()).orElse("file");
String scheme = Optional.ofNullable(uri.getScheme()).orElse("file");

switch (scheme) {
case "http":
Expand All @@ -45,7 +67,24 @@ public static void fetchFileFromUri(
break;

case "file":
Files.createSymbolicLink(destFile.toPath(), new File(uri.getPath()).toPath());
var srcPath = new File(uri.getPath()).toPath().normalize();
var destPath = destFile.toPath().toAbsolutePath().normalize();
Object lock = SYMLINK_LOCKS.computeIfAbsent(destPath.toString(), k -> new Object());
synchronized (lock) {
// Skip if the symlink already points to the correct target.
if (Files.isSymbolicLink(destPath)
&& Files.readSymbolicLink(destPath).normalize().equals(srcPath)) {
break;
}
// Replace via a temporary symlink + rename to minimize the window where the
// keytab path is absent (which could cause loginUserFromKeytab to fail).
// REPLACE_EXISTING is used here; on common local filesystems (ext4, xfs, APFS)
// a same-directory rename is effectively atomic at the OS level.
var tmpPath = destPath.resolveSibling(destPath.getFileName() + ".symlink.tmp");
Files.deleteIfExists(tmpPath);
Files.createSymbolicLink(tmpPath, srcPath);
Files.move(tmpPath, destPath, StandardCopyOption.REPLACE_EXISTING);
}
break;

case "hdfs":
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -172,10 +172,6 @@ public File saveKeyTabFileFromUri(String path) throws IOException {
keytabsDir.mkdir();
}
File keytabFile = new File(path);
if (keytabFile.exists() && !keytabFile.delete()) {
throw new IllegalStateException(
String.format("Fail to delete keytab file %s", keytabFile.getAbsolutePath()));
}
int fetchKeytabFileTimeout = kerberosConfig.getFetchTimeoutSec();
FetchFileUtils.fetchFileFromUri(keyTabUri, keytabFile, fetchKeytabFileTimeout, hadoopConf);
return keytabFile;
Expand All @@ -193,6 +189,7 @@ public void close() {
}

Files.deleteIfExists(Paths.get(keytabFilePath));
FetchFileUtils.removeLock(new File(keytabFilePath));
} catch (IOException e) {
LOG.warn("Failed to delete keytab file: {}", keytabFilePath, e);
}
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,119 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.gravitino.hive.kerberos;

import java.io.File;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import org.apache.hadoop.conf.Configuration;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;

public class TestFetchFileUtils {

@TempDir File tempDir;

@Test
public void testLinkLocalFile() throws Exception {
File srcFile = new File(tempDir, "source");
Assertions.assertTrue(srcFile.createNewFile());
File destFile = new File(tempDir, "dest");

FetchFileUtils.fetchFileFromUri(srcFile.toURI().toString(), destFile, 10, new Configuration());
Assertions.assertTrue(Files.isSymbolicLink(destFile.toPath()));
Assertions.assertEquals(srcFile.toPath(), Files.readSymbolicLink(destFile.toPath()));
}

@Test
public void testConcurrentSymlinkCreation() throws Exception {
File srcFile = new File(tempDir, "source_concurrent");
Assertions.assertTrue(srcFile.createNewFile());
File destFile = new File(tempDir, "dest_concurrent");

int threadCount = 10;
CyclicBarrier barrier = new CyclicBarrier(threadCount);
ExecutorService executor = Executors.newFixedThreadPool(threadCount);
List<Future<?>> futures = new ArrayList<>();
Comment thread
diqiu50 marked this conversation as resolved.
try {
for (int i = 0; i < threadCount; i++) {
futures.add(
executor.submit(
() -> {
try {
barrier.await(30, TimeUnit.SECONDS);
FetchFileUtils.fetchFileFromUri(
srcFile.toURI().toString(), destFile, 10, new Configuration());
Comment thread
diqiu50 marked this conversation as resolved.
} catch (Exception e) {
throw new RuntimeException(e);
}
}));
}
for (Future<?> future : futures) {
future.get(30, TimeUnit.SECONDS);
}
Comment thread
diqiu50 marked this conversation as resolved.
} finally {
executor.shutdownNow();
}

Assertions.assertTrue(Files.isSymbolicLink(destFile.toPath()));
Assertions.assertEquals(srcFile.toPath(), Files.readSymbolicLink(destFile.toPath()));
}

@Test
public void testIdempotentSymlinkCreation() throws Exception {
File srcFile = new File(tempDir, "source_idempotent");
Assertions.assertTrue(srcFile.createNewFile());
File destFile = new File(tempDir, "dest_idempotent");

Configuration conf = new Configuration();
String uri = srcFile.toURI().toString();

FetchFileUtils.fetchFileFromUri(uri, destFile, 10, conf);
Assertions.assertTrue(Files.isSymbolicLink(destFile.toPath()));

// Second call to the same dest should succeed without error
FetchFileUtils.fetchFileFromUri(uri, destFile, 10, conf);
Assertions.assertTrue(Files.isSymbolicLink(destFile.toPath()));
Assertions.assertEquals(srcFile.toPath(), Files.readSymbolicLink(destFile.toPath()));
}

@Test
public void testSymlinkReplacedWithDifferentTarget() throws Exception {
File srcFileA = new File(tempDir, "source_a");
File srcFileB = new File(tempDir, "source_b");
Assertions.assertTrue(srcFileA.createNewFile());
Assertions.assertTrue(srcFileB.createNewFile());
File destFile = new File(tempDir, "dest_replace");

Configuration conf = new Configuration();

FetchFileUtils.fetchFileFromUri(srcFileA.toURI().toString(), destFile, 10, conf);
Assertions.assertEquals(srcFileA.toPath(), Files.readSymbolicLink(destFile.toPath()));

FetchFileUtils.fetchFileFromUri(srcFileB.toURI().toString(), destFile, 10, conf);
Assertions.assertEquals(srcFileB.toPath(), Files.readSymbolicLink(destFile.toPath()));
}
}
Loading