This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 4794696325 [Cherry-pick to branch-1.3] [#11939] fix(core,auth):
tolerate concurrent directory creation instead of failing spuriously (#11940)
(#11945)
4794696325 is described below
commit 47946963250656c794bf516d7a7e992220366fc5
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Jul 9 11:30:37 2026 +0800
[Cherry-pick to branch-1.3] [#11939] fix(core,auth): tolerate concurrent
directory creation instead of failing spuriously (#11940) (#11945)
**Cherry-pick Information:**
- Original commit: 82fc98bddc70748d9166caff224ed52da3c61436
- Target branch: `branch-1.3`
- Status: ✅ **Conflicts resolved**
**Conflict resolution note:**
This backport intentionally keeps only the `JobManager`
staging-directory fix
(`core/.../job/JobManager.java` + `TestJobManager.java`).
The keytab (`KerberosAuthUtils`) half of #11940 does **not** apply to
`branch-1.3`:
the shared `catalogs/hadoop-auth` module was introduced by #11765, which
is not
present on `branch-1.3`. On `branch-1.3` the keytab code still lives in
the
per-catalog `KerberosClient` classes and already ignores the `mkdir()`
return
value, so the concurrent-creation bug fixed upstream never existed
there. The
GitHub auto-cherry-pick had added `KerberosAuthUtils.java` as new files
under a
module that isn't in `settings.gradle.kts`; those orphan files were
removed.
Verified: `./gradlew :core:test --tests TestJobManager` — 17/17 passing,
including the new concurrent staging-directory test.
Co-authored-by: Qi Yu <[email protected]>
---
.../java/org/apache/gravitino/job/JobManager.java | 14 ++++++---
.../org/apache/gravitino/job/TestJobManager.java | 35 ++++++++++++++++++++++
2 files changed, 45 insertions(+), 4 deletions(-)
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
