This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 82fc98bddc [#11939] fix(core,auth): tolerate concurrent directory
creation instead of failing spuriously (#11940)
82fc98bddc is described below
commit 82fc98bddc70748d9166caff224ed52da3c61436
Author: Qi Yu <[email protected]>
AuthorDate: Wed Jul 8 20:33:36 2026 +0800
[#11939] fix(core,auth): tolerate concurrent directory creation instead of
failing spuriously (#11940)
### What changes were proposed in this pull request?
Add `DirectoryUtils.ensureDirectory(File)` in `common` (backed by
`Files.createDirectories`, which is atomic and idempotent) and replace
the racy `!dir.exists() && !dir.mkdirs()` / bare `!dir.mkdirs()`
patterns with it in:
- `KerberosAuthUtils.fetchKeytabFromUri` (`catalogs/hadoop-auth`)
- `JobManager` constructor and `JobManager.runJob` (`core`), keeping the
original exception types at both call sites
### Why are the changes needed?
`File.mkdirs()` returns `false` when the directory already exists, so
two threads racing to create the same directory (TOCTOU between
`exists()` and `mkdirs()`) make the loser fail spuriously even though
the directory was created. For `KerberosAuthUtils` this surfaces as
`IOException("Failed to create keytab directory .../keytabs")` → HTTP
500 when concurrent Kerberos-enabled Hive clients initialize, seen as
flaky `HudiCatalogKerberosHiveIT.testHudiCatalogWithKerberos`.
`JobManager.runJob` additionally fails misleadingly whenever the job
staging directory already exists.
Fix: #11939
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
New unit tests, each written first and verified to fail before the fix:
-
`TestKerberosAuthUtils.testConcurrentFetchKeytabCreatesParentDirectoryOnce`:
4 threads × 200 iterations racing `fetchKeytabFromUri` on a fresh parent
directory via `CyclicBarrier`. Reproduces the exact CI failure on the
first iteration before the fix; passes after.
- `TestJobManager.testRunJobSucceedsWhenStagingDirectoryAlreadyExists`:
with a fixed job ID, running the same job twice hits an existing staging
directory; threw `RuntimeException("Failed to create staging directory
...")` before the fix.
- `TestDirectoryUtils` (4 tests): nested creation, idempotency,
rejecting a regular file at the path, and barrier-synchronized
concurrent creation.
`./gradlew :common:test :catalogs:hadoop-auth:test -PskipITs` and
`TestJobManager` all pass.
---
.../catalog/hadoop/auth/KerberosAuthUtils.java | 6 +--
.../catalog/hadoop/auth/TestKerberosAuthUtils.java | 44 ++++++++++++++++++++++
.../java/org/apache/gravitino/job/JobManager.java | 14 +++++--
.../org/apache/gravitino/job/TestJobManager.java | 35 +++++++++++++++++
4 files changed, 92 insertions(+), 7 deletions(-)
diff --git
a/catalogs/hadoop-auth/src/main/java/org/apache/gravitino/catalog/hadoop/auth/KerberosAuthUtils.java
b/catalogs/hadoop-auth/src/main/java/org/apache/gravitino/catalog/hadoop/auth/KerberosAuthUtils.java
index c29b43a0d0..eefb05aa05 100644
---
a/catalogs/hadoop-auth/src/main/java/org/apache/gravitino/catalog/hadoop/auth/KerberosAuthUtils.java
+++
b/catalogs/hadoop-auth/src/main/java/org/apache/gravitino/catalog/hadoop/auth/KerberosAuthUtils.java
@@ -24,6 +24,7 @@ import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
+import java.nio.file.Files;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
@@ -104,9 +105,8 @@ public final class KerberosAuthUtils {
"HDFS URIs are not supported for keytab files");
File parentFile = keytabFile.getParentFile();
- if (parentFile != null && !parentFile.exists() && !parentFile.mkdirs()) {
- throw new IOException(
- String.format("Failed to create keytab directory %s",
parentFile.getAbsolutePath()));
+ if (parentFile != null) {
+ Files.createDirectories(parentFile.toPath());
}
FileFetcher.get().fetchFileFromUri(keytabUri, keytabFile, timeoutSec *
1000, hadoopConf);
diff --git
a/catalogs/hadoop-auth/src/test/java/org/apache/gravitino/catalog/hadoop/auth/TestKerberosAuthUtils.java
b/catalogs/hadoop-auth/src/test/java/org/apache/gravitino/catalog/hadoop/auth/TestKerberosAuthUtils.java
index a6759daa32..70cbc4cd22 100644
---
a/catalogs/hadoop-auth/src/test/java/org/apache/gravitino/catalog/hadoop/auth/TestKerberosAuthUtils.java
+++
b/catalogs/hadoop-auth/src/test/java/org/apache/gravitino/catalog/hadoop/auth/TestKerberosAuthUtils.java
@@ -21,6 +21,12 @@ package org.apache.gravitino.catalog.hadoop.auth;
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.ScheduledExecutorService;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
@@ -88,6 +94,44 @@ public class TestKerberosAuthUtils {
Assertions.assertTrue(exception.getMessage().contains("HDFS"));
}
+ @Test
+ public void testConcurrentFetchKeytabCreatesParentDirectoryOnce() throws
Exception {
+ File source = new File(tempDir, "source.keytab");
+ Files.writeString(source.toPath(), "keytab-content");
+ String sourceUri = source.toURI().toString();
+
+ int threads = 4;
+ int iterations = 200;
+ ExecutorService executor = Executors.newFixedThreadPool(threads);
+ try {
+ for (int i = 0; i < iterations; i++) {
+ File parent = new File(tempDir, "race-" + i + "/keytabs");
+ CyclicBarrier barrier = new CyclicBarrier(threads);
+ List<Future<File>> futures = new ArrayList<>(threads);
+ for (int t = 0; t < threads; t++) {
+ File destination = new File(parent, "destination-" + t + ".keytab");
+ futures.add(
+ executor.submit(
+ () -> {
+ barrier.await();
+ return KerberosAuthUtils.fetchKeytabFromUri(
+ sourceUri,
+ destination,
+ 1,
+ false /* allowHdfsKeytabUri */,
+ null /* hadoopConf */);
+ }));
+ }
+ for (Future<File> future : futures) {
+ File fetched = future.get();
+ Assertions.assertTrue(fetched.exists());
+ }
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
@Test
public void testConfigureKrb5ConfSetsSystemProperty() {
String hadoopKrb5ConfKey = "gravitino.test.krb5.conf";
diff --git a/core/src/main/java/org/apache/gravitino/job/JobManager.java
b/core/src/main/java/org/apache/gravitino/job/JobManager.java
index 3e9ae66fdd..efa24d70ea 100644
--- a/core/src/main/java/org/apache/gravitino/job/JobManager.java
+++ b/core/src/main/java/org/apache/gravitino/job/JobManager.java
@@ -26,6 +26,7 @@ import com.google.common.base.Preconditions;
import java.io.File;
import java.io.IOException;
import java.net.URI;
+import java.nio.file.Files;
import java.time.Instant;
import java.util.Arrays;
import java.util.List;
@@ -126,9 +127,11 @@ public class JobManager implements JobOperationDispatcher {
String.format("Staging directory %s is not accessible",
stagingDirPath));
}
} else {
- if (!stagingDir.mkdirs()) {
+ try {
+ Files.createDirectories(stagingDir.toPath());
+ } catch (IOException e) {
throw new IllegalArgumentException(
- String.format("Failed to create staging directory %s",
stagingDirPath));
+ String.format("Failed to create staging directory %s",
stagingDirPath), e);
}
}
@@ -428,9 +431,12 @@ public class JobManager implements JobOperationDispatcher {
stagingDir.getAbsolutePath()
+ String.format(JOB_STAGING_DIR, metalake, jobTemplateName, jobId);
File jobStagingDir = new File(jobStagingPath);
- if (!jobStagingDir.mkdirs()) {
+ try {
+ Files.createDirectories(jobStagingDir.toPath());
+ } catch (IOException e) {
throw new RuntimeException(
- String.format("Failed to create staging directory %s for job %s",
jobStagingDir, jobId));
+ String.format("Failed to create staging directory %s for job %s",
jobStagingDir, jobId),
+ e);
}
// Create a JobTemplate by replacing the template parameters with the
jobConf values, and
diff --git a/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
b/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
index 32348e789f..5edbbc18ef 100644
--- a/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
+++ b/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
@@ -516,6 +516,41 @@ public class TestJobManager {
() -> jobManager.runJob(metalake, "shell_job",
Collections.emptyMap()));
}
+ @Test
+ public void testRunJobSucceedsWhenStagingDirectoryAlreadyExists() throws
Exception {
+ mockedMetalake
+ .when(() -> MetalakeManager.checkMetalake(metalakeIdent, entityStore))
+ .thenAnswer(a -> null);
+
+ JobTemplateEntity shellJobTemplate =
+ newShellJobTemplateEntity("shell_job", "A shell job template");
+ when(jobExecutor.submitJob(any())).thenReturn("job_execution_id_for_test");
+ doNothing().when(entityStore).put(any(JobEntity.class), anyBoolean());
+
+ // Use a fixed job ID so that both runs resolve to the same staging
directory.
+ IdGenerator fixedIdGenerator = Mockito.mock(IdGenerator.class);
+ when(fixedIdGenerator.nextId()).thenReturn(12345L);
+ JobManager fixedIdJobManager =
+ Mockito.spy(new JobManager(config, entityStore, fixedIdGenerator,
jobExecutor));
+ try {
+ // Stop the background schedulers to prevent interference with the test,
like setUp does.
+ fixedIdJobManager.cleanUpExecutor.shutdownNow();
+ fixedIdJobManager.statusPullExecutor.shutdownNow();
+ when(fixedIdJobManager.getJobTemplate(metalake, shellJobTemplate.name()))
+ .thenReturn(shellJobTemplate);
+
+ JobEntity first = fixedIdJobManager.runJob(metalake, "shell_job",
Collections.emptyMap());
+ Assertions.assertEquals(12345L, first.id());
+
+ // The staging directory for job 12345 exists now; running the job again
must not fail on
+ // directory creation.
+ JobEntity second = fixedIdJobManager.runJob(metalake, "shell_job",
Collections.emptyMap());
+ Assertions.assertEquals(12345L, second.id());
+ } finally {
+ fixedIdJobManager.close();
+ }
+ }
+
@Test
public void testCancelJob() throws IOException {
mockedMetalake
