[#10741] fix(hive): Fix keytab symlink TOCTOU race in FetchFileUtils (#10742)

### What changes were proposed in this pull request?

Fix the `file:` scheme handler in `FetchFileUtils` to synchronize
symlink creation and avoid concurrent races. Replace the non-atomic
`exists() + delete()` in `KerberosClient` with `Files.deleteIfExists()`.
Remove the unused duplicate `FetchFileUtils` in `catalog-hive` and add
comprehensive tests in `hive-metastore-common`.

### Why are the changes needed?

`Files.createSymbolicLink()` throws `FileAlreadyExistsException` when
multiple threads concurrently fetch the same keytab path, such as when
the search event listener triggers cascade sync after catalog creation.

Fix: #10741

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

- added unit coverage in `TestFetchFileUtils` including concurrent,
idempotent, and target-replacement scenarios

Author: diqiu50

catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/FetchFileUtils.java

@@ -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<>();
+
   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":
@@ -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":

catalogs/catalog-hive/src/test/java/org/apache/gravitino/catalog/hive/TestFetchFileUtils.java

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

catalogs/hive-metastore-common/src/main/java/org/apache/gravitino/hive/kerberos/FetchFileUtils.java

@@ -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<>();
+    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());
+                  } catch (Exception e) {
+                    throw new RuntimeException(e);
+                  }
+                }));
+      }
+      for (Future<?> future : futures) {
+        future.get(30, TimeUnit.SECONDS);
+      }
+    } 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()));
+  }
+}