This is an automated email from the ASF dual-hosted git repository.
jerryshao 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 2d63db89d9 [#13678] fix(core): make a cancelled local job end as
CANCELLED (#13679)
2d63db89d9 is described below
commit 2d63db89d926b104b03abbb71918ec2a7cbf5aa0
Author: YangJie <[email protected]>
AuthorDate: Wed Oct 7 23:34:52 2026 -0400
[#13678] fix(core): make a cancelled local job end as CANCELLED (#13679)
### What changes were proposed in this pull request?
`LocalJobExecutor.cancelJob` still sends SIGTERM first, and now also
schedules a check that kills the process forcibly if it is still alive
after `gravitino.jobExecutor.local.cancelForceKillDelayInMs` (30 seconds
by default). `runJob` now checks whether the job was being cancelled
before it looks at the exit code, so a cancelled job ends CANCELLED
whatever code its process exits with. The new setting is added to the
local executor configuration table in
`docs/manage-jobs-in-gravitino.md`.
Like the existing SIGTERM, the forced kill reaches only the job's own
process, not processes it started.
### Why are the changes needed?
A cancelled job whose process ignored SIGTERM never exited: it stayed
CANCELLING and kept its worker thread blocked in `waitFor()`. The worker
pool is fixed at `maxRunningJobs`, so each such job took a slot for
good, and once all slots were held new jobs stayed queued until a
restart. Separately, a cancelled job whose process trapped SIGTERM and
exited 0 was reported SUCCEEDED, because the exit code was checked
first.
Fix: #13678
### Does this PR introduce _any_ user-facing change?
Yes. A new setting,
`gravitino.jobExecutor.local.cancelForceKillDelayInMs` (default
`30000`), controls how long a cancelled job's process may keep running
before it is killed forcibly. A cancelled job that exits 0 is now
reported CANCELLED instead of SUCCEEDED.
### How was this patch tested?
Added `testCancelJobEscalatesToForcibleKill`, which cancels a job that
ignores SIGTERM and asserts it reaches CANCELLED once the force-kill
delay passes, and `testCancelledJobExitingZeroReportsCancelled`, which
cancels a job that traps SIGTERM and exits 0 and asserts CANCELLED. Each
script prints a marker after installing its trap, and the test waits for
it before cancelling so the signal cannot arrive first. Both tests fail
against the code before this change.
---
.../gravitino/job/local/LocalJobExecutor.java | 54 ++++++++---
.../job/local/LocalJobExecutorConfigs.java | 9 ++
.../gravitino/job/local/TestLocalJobExecutor.java | 103 +++++++++++++++++++++
docs/manage-jobs-in-gravitino.md | 13 +--
4 files changed, 159 insertions(+), 20 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/job/local/LocalJobExecutor.java
b/core/src/main/java/org/apache/gravitino/job/local/LocalJobExecutor.java
index a8f83702c8..9c42bad643 100644
--- a/core/src/main/java/org/apache/gravitino/job/local/LocalJobExecutor.java
+++ b/core/src/main/java/org/apache/gravitino/job/local/LocalJobExecutor.java
@@ -19,6 +19,8 @@
package org.apache.gravitino.job.local;
+import static
org.apache.gravitino.job.local.LocalJobExecutorConfigs.CANCEL_FORCE_KILL_DELAY_MS;
+import static
org.apache.gravitino.job.local.LocalJobExecutorConfigs.DEFAULT_CANCEL_FORCE_KILL_DELAY_MS;
import static
org.apache.gravitino.job.local.LocalJobExecutorConfigs.DEFAULT_JOB_STATUS_KEEP_TIME_MS;
import static
org.apache.gravitino.job.local.LocalJobExecutorConfigs.DEFAULT_MAX_RUNNING_JOBS;
import static
org.apache.gravitino.job.local.LocalJobExecutorConfigs.DEFAULT_WAITING_QUEUE_SIZE;
@@ -143,6 +145,8 @@ public class LocalJobExecutor implements JobExecutor {
private Map<String, JobExecutionInfo> jobInfos;
private final Object lock = new Object();
+ private long cancelForceKillDelayMs;
+
private long jobStatusKeepTimeInMs;
private ScheduledExecutorService jobStatusCleanupExecutor;
@@ -169,6 +173,14 @@ public class LocalJobExecutor implements JobExecutor {
@Override
public void initialize(Map<String, String> configs) {
this.configs = configs;
+ this.cancelForceKillDelayMs =
+ configs.containsKey(CANCEL_FORCE_KILL_DELAY_MS)
+ ? Long.parseLong(configs.get(CANCEL_FORCE_KILL_DELAY_MS))
+ : DEFAULT_CANCEL_FORCE_KILL_DELAY_MS;
+ Preconditions.checkArgument(
+ cancelForceKillDelayMs > 0,
+ "Cancel force-kill delay must be greater than 0, but got: %s",
+ cancelForceKillDelayMs);
this.executorId = String.format("%08x",
ThreadLocalRandom.current().nextInt());
this.ownedJobIdPrefix = LOCAL_JOB_PREFIX + executorId + "-";
LOG.info("Initializing local job executor with executor id {}",
executorId);
@@ -366,6 +378,20 @@ public class LocalJobExecutor implements JobExecutor {
Process process = runningProcesses.get(jobId);
if (process != null) {
process.destroy();
+ // A process that traps or ignores SIGTERM would otherwise stay in
+ // CANCELLING forever and permanently occupy one of the fixed worker
+ // threads; escalate to a hard kill after a grace period.
+ long delayMs = cancelForceKillDelayMs;
+ jobStatusCleanupExecutor.schedule(
+ () -> {
+ if (process.isAlive()) {
+ LOG.warn(
+ "Job {} still alive {} ms after cancel, destroying
forcibly", jobId, delayMs);
+ process.destroyForcibly();
+ }
+ },
+ delayMs,
+ TimeUnit.MILLISECONDS);
}
LOG.info("Job {} is cancelling while running", jobId);
jobInfos.put(jobId,
info.toBuilder().withStatus(JobHandle.Status.CANCELLING).build());
@@ -446,21 +472,21 @@ public class LocalJobExecutor implements JobExecutor {
LOG.info("Starting job: {}", jobId);
int exitCode = process.waitFor();
- if (exitCode == 0) {
- LOG.info("Job {} completed successfully", jobId);
- synchronized (lock) {
+ synchronized (lock) {
+ // The job may be missing if the executor is closed concurrently.
+ JobExecutionInfo current = jobInfos.get(jobId);
+ JobHandle.Status oldStatus = current == null ? null : current.status();
+ if (oldStatus == JobHandle.Status.CANCELLING) {
+ // A cancelled job whose process traps SIGTERM and still exits 0 was
cancelled, not
+ // successful; consult the cancel state before the exit code.
+ LOG.info("Job {} was cancelled while running with exit code: {}",
jobId, exitCode);
+ finishJob(jobId, JobHandle.Status.CANCELLED);
+ } else if (exitCode == 0) {
+ LOG.info("Job {} completed successfully", jobId);
finishJob(jobId, JobHandle.Status.SUCCEEDED);
- }
- } else {
- synchronized (lock) {
- JobHandle.Status oldStatus = jobInfos.get(jobId).status();
- if (oldStatus == JobHandle.Status.CANCELLING) {
- LOG.info("Job {} was cancelled while running with exit code: {}",
jobId, exitCode);
- finishJob(jobId, JobHandle.Status.CANCELLED);
- } else if (oldStatus == JobHandle.Status.STARTED) {
- LOG.warn("Job {} failed after starting with exit code: {}", jobId,
exitCode);
- finishJob(jobId, JobHandle.Status.FAILED);
- }
+ } else if (oldStatus == null || oldStatus == JobHandle.Status.STARTED)
{
+ LOG.warn("Job {} failed after starting with exit code: {}", jobId,
exitCode);
+ finishJob(jobId, JobHandle.Status.FAILED);
}
}
diff --git
a/core/src/main/java/org/apache/gravitino/job/local/LocalJobExecutorConfigs.java
b/core/src/main/java/org/apache/gravitino/job/local/LocalJobExecutorConfigs.java
index c02eddb902..10c055cb08 100644
---
a/core/src/main/java/org/apache/gravitino/job/local/LocalJobExecutorConfigs.java
+++
b/core/src/main/java/org/apache/gravitino/job/local/LocalJobExecutorConfigs.java
@@ -36,6 +36,15 @@ public class LocalJobExecutorConfigs {
public static final String JOB_STATUS_KEEP_TIME_MS = "jobStatusKeepTimeInMs";
public static final long DEFAULT_JOB_STATUS_KEEP_TIME_MS = 60 * 60 * 1000;
// 1 hour
+ /**
+ * How long in milliseconds a cancelled job's process may keep running after
it is asked to stop
+ * before the executor kills it forcibly. Must be greater than 0.
+ */
+ public static final String CANCEL_FORCE_KILL_DELAY_MS =
"cancelForceKillDelayInMs";
+
+ /** The default value of {@link #CANCEL_FORCE_KILL_DELAY_MS}. */
+ public static final long DEFAULT_CANCEL_FORCE_KILL_DELAY_MS = 30 * 1000L; //
30 seconds
+
public static final String SPARK_HOME = "sparkHome";
/**
diff --git
a/core/src/test/java/org/apache/gravitino/job/local/TestLocalJobExecutor.java
b/core/src/test/java/org/apache/gravitino/job/local/TestLocalJobExecutor.java
index 8f1676ef42..0fd29d9d46 100644
---
a/core/src/test/java/org/apache/gravitino/job/local/TestLocalJobExecutor.java
+++
b/core/src/test/java/org/apache/gravitino/job/local/TestLocalJobExecutor.java
@@ -639,6 +639,96 @@ public class TestLocalJobExecutor {
Assertions.assertEquals(JobHandle.Status.CANCELLED,
jobExecutor.getJobStatus(jobId));
}
+ @Test
+ public void testCancelJobEscalatesToForcibleKill() throws Exception {
+ // A process that ignores SIGTERM must still terminate (and free its worker
+ // thread) after the cancel grace period.
+ LocalJobExecutor executor = new LocalJobExecutor();
+ executor.initialize(
+ withStagingDir(
+
ImmutableMap.of(LocalJobExecutorConfigs.CANCEL_FORCE_KILL_DELAY_MS, "1000")));
+ try {
+ JobTemplateEntity.TemplateContent trapContent =
+ JobTemplateEntity.TemplateContent.builder()
+ .withExecutable("/bin/sh")
+ .withArguments(Lists.newArrayList("-c", "trap '' TERM; echo
ready; exec sleep 300"))
+ .withEnvironments(ImmutableMap.of())
+ .withJobType(JobTemplate.JobType.SHELL)
+ .withScripts(Lists.newArrayList())
+ .withCustomFields(ImmutableMap.of())
+ .build();
+ JobTemplateEntity trapTemplate =
+ JobTemplateEntity.builder()
+ .withId(2L)
+ .withName("trap-term-job-template")
+ .withNamespace(NamespaceUtil.ofJobTemplate("test"))
+ .withComment("test")
+ .withTemplateContent(trapContent)
+ .withAuditInfo(AuditInfo.EMPTY)
+ .build();
+ JobTemplate template =
+ new JobTemplateResolver(trapTemplate).resolve(ImmutableMap.of(),
workingDir);
+
+ String jobId = executor.submitJob(template);
+ awaitReadyMarker();
+ Assertions.assertEquals(JobHandle.Status.STARTED,
executor.getJobStatus(jobId));
+
+ executor.cancelJob(jobId);
+
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .until(() -> executor.getJobStatus(jobId) ==
JobHandle.Status.CANCELLED);
+ } finally {
+ executor.close();
+ }
+ }
+
+ @Test
+ public void testCancelledJobExitingZeroReportsCancelled() throws Exception {
+ // A cancelled job whose TERM handler exits 0 must report CANCELLED, not
SUCCEEDED: the user
+ // asked for it to stop, so a clean exit taken in response to the cancel
is still a
+ // cancellation.
+ LocalJobExecutor executor = new LocalJobExecutor();
+ executor.initialize(withStagingDir(Collections.emptyMap()));
+ try {
+ JobTemplateEntity.TemplateContent content =
+ JobTemplateEntity.TemplateContent.builder()
+ .withExecutable("/bin/sh")
+ .withArguments(
+ Lists.newArrayList(
+ "-c", "trap 'exit 0' TERM; echo ready; while :; do sleep
1; done"))
+ .withEnvironments(ImmutableMap.of())
+ .withJobType(JobTemplate.JobType.SHELL)
+ .withScripts(Lists.newArrayList())
+ .withCustomFields(ImmutableMap.of())
+ .build();
+ JobTemplateEntity templateEntity =
+ JobTemplateEntity.builder()
+ .withId(3L)
+ .withName("term-exit-zero-template")
+ .withNamespace(NamespaceUtil.ofJobTemplate("test"))
+ .withComment("test")
+ .withTemplateContent(content)
+ .withAuditInfo(AuditInfo.EMPTY)
+ .build();
+ JobTemplate template =
+ new JobTemplateResolver(templateEntity).resolve(ImmutableMap.of(),
workingDir);
+
+ String jobId = executor.submitJob(template);
+ awaitReadyMarker();
+ Assertions.assertEquals(JobHandle.Status.STARTED,
executor.getJobStatus(jobId));
+
+ executor.cancelJob(jobId);
+
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .until(() -> executor.getJobStatus(jobId) ==
JobHandle.Status.CANCELLED);
+ Assertions.assertEquals(JobHandle.Status.CANCELLED,
executor.getJobStatus(jobId));
+ } finally {
+ executor.close();
+ }
+ }
+
@Test
public void testCancelSucceededJob() {
// Cancelling a job that is already succeeded.
@@ -1135,6 +1225,19 @@ public class TestLocalJobExecutor {
Assertions.assertEquals(6, jobExecutor.getJobStdout(jobId, 100,
DEFAULT_TEST_MAX_BYTES).size());
}
+ // The job script prints "ready" only after installing its TERM trap, so a
cancel sent once this
+ // returns cannot reach the shell before the trap is in place.
+ private static void awaitReadyMarker() {
+ File output = new File(workingDir, "output.log");
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .until(
+ () ->
+ output.exists()
+ && FileUtils.readFileToString(output,
StandardCharsets.UTF_8)
+ .contains("ready"));
+ }
+
private static Map<String, String> withStagingDir(Map<String, String>
configs) {
return ImmutableMap.<String, String>builder()
.putAll(configs)
diff --git a/docs/manage-jobs-in-gravitino.md b/docs/manage-jobs-in-gravitino.md
index 139f48d619..21ba58a285 100644
--- a/docs/manage-jobs-in-gravitino.md
+++ b/docs/manage-jobs-in-gravitino.md
@@ -384,12 +384,13 @@ default configurations:
The local job executor is used for testing and development purposes, it runs
the job in the local process.
The following are the default configurations for the local job executor:
-| Property name | Description
| Default value
| Required |
-|-----------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------|----------------------------------------|----------|
-| `gravitino.jobExecutor.local.waitingQueueSize` | The size of the
waiting queue for queued jobs in the local job executor
| `100`
| No |
-| `gravitino.jobExecutor.local.maxRunningJobs` | The maximum number of
running jobs in the local job executor
| `max(1, min(available cores / 2,
10))` | No |
-| `gravitino.jobExecutor.local.jobStatusKeepTimeInMs` | The time in
milliseconds to keep the job status in the local job executor
| `3600000` (1 hour)
| No |
-| `gravitino.jobExecutor.local.sparkHome` | The home directory of
Spark, Gravitino checks this configuration firstly and then `SPARK_HOME` env.
Either of them should be set to run Spark job | `None`
| No |
+| Property name | Description
| Default value | Required |
+|--------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------------------------------------|----------|
+| `gravitino.jobExecutor.local.waitingQueueSize` | The size of the
waiting queue for queued jobs in the local job executor
| `100` | No |
+| `gravitino.jobExecutor.local.maxRunningJobs` | The maximum number
of running jobs in the local job executor
| `max(1, min(available cores / 2, 10))` | No |
+| `gravitino.jobExecutor.local.jobStatusKeepTimeInMs` | The time in
milliseconds to keep the job status in the local job executor
| `3600000` (1 hour) | No |
+| `gravitino.jobExecutor.local.cancelForceKillDelayInMs` | How long in
milliseconds a cancelled job's process may keep running after it is asked to
stop before the executor kills it forcibly. Raise it for jobs that need longer
to shut down cleanly | `30000` (30 seconds) | No |
+| `gravitino.jobExecutor.local.sparkHome` | The home directory
of Spark, Gravitino checks this configuration firstly and then `SPARK_HOME`
env. Either of them should be set to run Spark job
| `None` | No |
The local job executor always uses `gravitino.job.stagingDir` as its staging
directory, the same one
the job system stages jobs in. A `gravitino.jobExecutor.local.stagingDir`
setting is ignored.
