You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by al...@apache.org on 2026/09/23 03:51:43 UTC
(asterixdb) branch master updated: [ASTERIXDB-3884][HYR] Release a job's state when its submission fails
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
The following commit(s) were added to refs/heads/master by this push:
new 01af06d2d0 [ASTERIXDB-3884][HYR] Release a job's state when its submission fails
01af06d2d0 is described below
commit 01af06d2d04694859e89f062659fcca199d09726
Author: Ali Alsuliman <al...@gmail.com>
AuthorDate: Tue Sep 22 13:09:28 2026 -0700
[ASTERIXDB-3884][HYR] Release a job's state when its submission fails
- user model changes: no
- storage format changes: no
- interface changes: yes
Details:
JobManager.add notifies the job lifecycle listeners that a job has been
created and only then queues or executes it. When that throws - a full
job queue being the reachable case - the job neither starts nor finishes
and its id never reaches the client, so no later notification and no
caller can release what the listeners registered. For an atomic
statement that strands the global transaction context, and the resource
map it holds, in a repository that is filled from the job lifecycle but
drained only from the statement path.
Notify the listeners of the failure instead, defaulting the new method
to a no-op so the other implementors are untouched. It is called from a
finally rather than the existing catch, which sees only
HyracksDataException while queueing, execution and an unknown submission
status can all fail unchecked.
Ext-ref: MB-73831
Co-Authored-By: Claude Opus 5 <no...@anthropic.com>
Change-Id: Ie110c960950a75c8edc7e5e5f92dc33b49e6115e
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21815
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Murtadha Hubail <mh...@apache.org>
---
.../org/apache/asterix/app/cc/GlobalTxManager.java | 6 ++
.../apache/asterix/app/cc/GlobalTxManagerTest.java | 31 +++++++
.../hyracks/api/job/IJobLifecycleListener.java | 14 ++++
.../control/cc/application/CCServiceContext.java | 16 ++++
.../apache/hyracks/control/cc/job/JobManager.java | 20 ++++-
.../hyracks/control/cc/job/JobManagerTest.java | 94 ++++++++++++++++++++++
6 files changed, 178 insertions(+), 3 deletions(-)
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/GlobalTxManager.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/GlobalTxManager.java
index 4a5e912529..17531ab1c9 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/GlobalTxManager.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/GlobalTxManager.java
@@ -314,6 +314,12 @@ public class GlobalTxManager implements IGlobalTxManager {
}
}
+ @Override
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.GENERATED)
+ public void notifyJobSubmissionFailed(JobId jobId, JobSpecification spec) {
+ txnContextRepository.remove(jobId);
+ }
+
@Override
public void notifyJobStart(JobId jobId, JobSpecification spec) throws HyracksException {
}
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/cc/GlobalTxManagerTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/cc/GlobalTxManagerTest.java
index d98d08143c..c9ec79743e 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/cc/GlobalTxManagerTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/cc/GlobalTxManagerTest.java
@@ -44,6 +44,7 @@ import org.apache.asterix.common.messaging.api.INcAddressedMessage;
import org.apache.asterix.common.transactions.IGlobalTransactionContext;
import org.apache.hyracks.api.application.ICCServiceContext;
import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.api.job.JobSpecification;
import org.apache.hyracks.control.cc.ClusterControllerService;
import org.apache.hyracks.control.common.controllers.CCConfig;
import org.apache.hyracks.control.nc.io.IOManager;
@@ -306,6 +307,36 @@ public class GlobalTxManagerTest {
rolledBack);
}
+ /**
+ * JobManager.add registers the transaction and only then queues or executes the job. When that throws, the
+ * job id never reaches the client, so the statement path has nothing to abort with and the context - and
+ * the resource map it holds - is retained for the lifetime of the cluster controller: the repository is an
+ * unbounded map filled from the job lifecycle and drained only from the statement path.
+ */
+ @Test
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.TEST_GENERATED, notes = "Covers deregistration of a transaction whose job was never submitted")
+ public void failedSubmissionDeregistersTheTransaction() {
+ JobId jobId = new JobId(8);
+ globalTxManager.beginTransaction(jobId, 1, 1, Collections.singletonList(DATASET_ID));
+ assertEquals("the transaction should be registered before the submission fails", 1, countTracked(jobId));
+
+ globalTxManager.notifyJobSubmissionFailed(jobId, mock(JobSpecification.class));
+
+ assertEquals("the transaction outlived the failed submission", 0, countTracked(jobId));
+ }
+
+ /**
+ * Every job's failed submission reaches every listener, so a job with no global transaction - which is
+ * most of them - has to pass through silently rather than fail looking up a context it never had.
+ */
+ @Test
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.TEST_GENERATED, notes = "Pins that a non-atomic job's failed submission is ignored")
+ public void failedSubmissionOfNonAtomicJobIsIgnored() throws Exception {
+ globalTxManager.notifyJobSubmissionFailed(new JobId(9), mock(JobSpecification.class));
+
+ verify(messageBroker, never()).sendRealTimeApplicationMessageToNC(any(INcAddressedMessage.class), anyString());
+ }
+
private IGlobalTransactionContext prepareTransaction(JobId jobId) {
IGlobalTransactionContext context =
globalTxManager.beginTransaction(jobId, 1, 1, Collections.singletonList(DATASET_ID));
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/IJobLifecycleListener.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/IJobLifecycleListener.java
index 6773dde29d..9f5ea203bb 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/IJobLifecycleListener.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/IJobLifecycleListener.java
@@ -22,6 +22,7 @@ import java.util.List;
import org.apache.hyracks.api.exceptions.HyracksException;
import org.apache.hyracks.api.job.resource.IJobCapacityController;
+import org.apache.hyracks.util.annotations.AiProvenance;
/**
* A listener for job related events
@@ -38,6 +39,19 @@ public interface IJobLifecycleListener {
void notifyJobCreation(JobId jobId, JobSpecification spec, IJobCapacityController.JobSubmissionStatus status)
throws HyracksException;
+ /**
+ * Notify the listener that a job it was told of in {@link #notifyJobCreation} will not run after all, its
+ * submission having failed. Neither {@link #notifyJobStart} nor {@link #notifyJobFinish} follows, and the
+ * job id never reaches the client, so whatever the creation notification registered is released here or
+ * not at all.
+ *
+ * @param jobId the job id
+ * @param spec the job specification
+ */
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.GENERATED)
+ default void notifyJobSubmissionFailed(JobId jobId, JobSpecification spec) {
+ }
+
/**
* Notify the listener that the job has started on the cluster controller
*
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/application/CCServiceContext.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/application/CCServiceContext.java
index 555d37e089..3264b6a5eb 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/application/CCServiceContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/application/CCServiceContext.java
@@ -43,8 +43,13 @@ import org.apache.hyracks.control.cc.ClusterControllerService;
import org.apache.hyracks.control.common.application.ServiceContext;
import org.apache.hyracks.control.common.context.ServerContext;
import org.apache.hyracks.control.common.utils.HyracksThreadFactory;
+import org.apache.hyracks.util.annotations.AiProvenance;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
public class CCServiceContext extends ServiceContext implements ICCServiceContext {
+
+ private static final Logger LOGGER = LogManager.getLogger();
private final ICCContext ccContext;
protected final Set<String> initPendingNodeIds;
@@ -100,6 +105,17 @@ public class CCServiceContext extends ServiceContext implements ICCServiceContex
}
}
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.GENERATED)
+ public synchronized void notifyJobSubmissionFailed(JobId jobId, JobSpecification spec) {
+ for (IJobLifecycleListener l : jobLifecycleListeners) {
+ try {
+ l.notifyJobSubmissionFailed(jobId, spec);
+ } catch (Exception e) {
+ LOGGER.error("failed to notify {} that the submission of {} failed", l, jobId, e);
+ }
+ }
+ }
+
@Override
public void addClusterLifecycleListener(IClusterLifecycleListener clusterLifecycleListener) {
clusterLifecycleListeners.add(clusterLifecycleListener);
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/job/JobManager.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/job/JobManager.java
index 269478a5bb..4f0a851bcb 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/job/JobManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/job/JobManager.java
@@ -57,6 +57,7 @@ import org.apache.hyracks.control.cc.scheduler.IJobQueue;
import org.apache.hyracks.control.common.controllers.CCConfig;
import org.apache.hyracks.control.common.work.IResultCallback;
import org.apache.hyracks.control.common.work.NoOpCallback;
+import org.apache.hyracks.util.annotations.AiProvenance;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -126,13 +127,19 @@ public class JobManager implements IJobManager {
}
@Override
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.REFACTORED)
public void add(JobRun jobRun) throws HyracksException {
checkJob(jobRun);
JobSpecification job = jobRun.getJobSpecification();
- IJobCapacityController.JobSubmissionStatus status;
+ CCServiceContext serviceCtx = ccs.getContext();
+ boolean creationNotified = false;
+ boolean submitted = false;
try {
- status = jobCapacityController.allocate(job, jobRun.getJobId(), jobRun.getFlags());
- CCServiceContext serviceCtx = ccs.getContext();
+ IJobCapacityController.JobSubmissionStatus status =
+ jobCapacityController.allocate(job, jobRun.getJobId(), jobRun.getFlags());
+ // set before the notification, not after: it fans out to several listeners, and one that throws
+ // partway leaves the listeners ahead of it holding state only this failure path will release
+ creationNotified = true;
serviceCtx.notifyJobCreation(jobRun.getJobId(), job, status);
switch (status) {
case QUEUE:
@@ -144,9 +151,16 @@ public class JobManager implements IJobManager {
default:
throw new IllegalStateException("unknown submission status: " + status);
}
+ submitted = true;
} catch (HyracksDataException ex) {
handleException(ex);
throw ex;
+ } finally {
+ if (creationNotified && !submitted) {
+ // the job will neither start nor finish, so no later notification releases what the creation
+ // notification registered, and the client never receives the job id to release it either.
+ serviceCtx.notifyJobSubmissionFailed(jobRun.getJobId(), job);
+ }
}
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/test/java/org/apache/hyracks/control/cc/job/JobManagerTest.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/test/java/org/apache/hyracks/control/cc/job/JobManagerTest.java
index 3e87bb3cf1..32cdca4c0c 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/test/java/org/apache/hyracks/control/cc/job/JobManagerTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/test/java/org/apache/hyracks/control/cc/job/JobManagerTest.java
@@ -21,7 +21,9 @@ package org.apache.hyracks.control.cc.job;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -51,6 +53,7 @@ import org.apache.hyracks.control.common.controllers.CCConfig;
import org.apache.hyracks.control.common.ipc.NodeControllerRemoteProxy;
import org.apache.hyracks.control.common.logs.LogFile;
import org.apache.hyracks.control.common.work.NoOpCallback;
+import org.apache.hyracks.util.annotations.AiProvenance;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -273,6 +276,88 @@ public class JobManagerTest {
verify(jobManager, times(0)).finalComplete(any());
}
+ /**
+ * A job whose submission fails after the listeners were told of its creation is never started, never
+ * finished and never reported to the client, so nothing else can release what they registered for it. The
+ * atomic statement protocol registers a global transaction there, and its repository is drained only by
+ * the statement that holds the job id, so a stranded entry is retained for the lifetime of the CC.
+ */
+ @Test
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.TEST_GENERATED, notes = "Covers the job-queue-full submission failure")
+ public void queueFullNotifiesCreationFailed() throws HyracksException {
+ IJobCapacityController jobCapacityController = mock(IJobCapacityController.class);
+ ClusterControllerService ccs = mockClusterControllerService(0);
+ IJobManager jobManager = new JobManager(ccConfig, ccs, jobCapacityController);
+
+ JobRun run = mockJobRun(1);
+ JobSpecification job = mock(JobSpecification.class);
+ when(run.getJobSpecification()).thenReturn(job);
+ when(jobCapacityController.allocate(job, run.getJobId(), none))
+ .thenReturn(IJobCapacityController.JobSubmissionStatus.QUEUE);
+
+ boolean jobQueueFull = false;
+ try {
+ jobManager.add(run);
+ } catch (HyracksException e) {
+ jobQueueFull = e.matches(ErrorCode.JOB_QUEUE_FULL);
+ }
+
+ Assert.assertTrue(jobQueueFull);
+ verify(ccs.getContext()).notifyJobSubmissionFailed(run.getJobId(), job);
+ }
+
+ /**
+ * The same, for a submission that fails unchecked. Queueing and execution can both throw a
+ * RuntimeException, and an unknown submission status throws IllegalStateException, none of which the
+ * HyracksDataException catch in {@code add} sees.
+ */
+ @Test
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.TEST_GENERATED, notes = "Covers an unchecked submission failure, which the HyracksDataException catch misses")
+ public void uncheckedSubmissionFailureNotifiesCreationFailed() throws HyracksException {
+ IJobCapacityController jobCapacityController = mock(IJobCapacityController.class);
+ ClusterControllerService ccs = mockClusterControllerService();
+ IJobManager jobManager = new JobManager(ccConfig, ccs, jobCapacityController);
+
+ JobRun run = mockJobRun(1);
+ JobSpecification job = mock(JobSpecification.class);
+ when(run.getJobSpecification()).thenReturn(job);
+ when(jobCapacityController.allocate(job, run.getJobId(), none))
+ .thenReturn(IJobCapacityController.JobSubmissionStatus.EXECUTE);
+ doThrow(new IllegalStateException("cannot start")).when(run).setStatus(eq(JobStatus.RUNNING), any());
+
+ boolean failed = false;
+ try {
+ jobManager.add(run);
+ } catch (IllegalStateException e) {
+ failed = true;
+ }
+
+ Assert.assertTrue(failed);
+ verify(ccs.getContext()).notifyJobSubmissionFailed(run.getJobId(), job);
+ }
+
+ /**
+ * The negative control: a job that is accepted is released by its completion, so notifying a failed
+ * creation for it would deregister a transaction the statement is still going to commit.
+ */
+ @Test
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.TEST_GENERATED, notes = "Pins that an accepted submission is not reported as failed")
+ public void acceptedSubmissionDoesNotNotifyCreationFailed() throws HyracksException {
+ IJobCapacityController jobCapacityController = mock(IJobCapacityController.class);
+ ClusterControllerService ccs = mockClusterControllerService();
+ IJobManager jobManager = new JobManager(ccConfig, ccs, jobCapacityController);
+
+ JobRun run = mockJobRun(1);
+ JobSpecification job = mock(JobSpecification.class);
+ when(run.getJobSpecification()).thenReturn(job);
+ when(jobCapacityController.allocate(job, run.getJobId(), none))
+ .thenReturn(IJobCapacityController.JobSubmissionStatus.EXECUTE);
+
+ jobManager.add(run);
+
+ verify(ccs.getContext(), never()).notifyJobSubmissionFailed(any(), any());
+ }
+
private JobRun mockJobRun(long id) {
JobRun run = mock(JobRun.class, Mockito.RETURNS_DEEP_STUBS);
when(run.getExceptions()).thenReturn(Collections.emptyList());
@@ -289,6 +374,15 @@ public class JobManagerTest {
return run;
}
+ /** @param jobQueueCapacity the capacity the job queue is built with, rather than the configured one */
+ private ClusterControllerService mockClusterControllerService(int jobQueueCapacity) {
+ ClusterControllerService ccs = mockClusterControllerService();
+ CCConfig queueConfig = mock(CCConfig.class);
+ when(queueConfig.getJobQueueCapacity()).thenReturn(jobQueueCapacity);
+ when(ccs.getCCConfig()).thenReturn(queueConfig);
+ return ccs;
+ }
+
private ClusterControllerService mockClusterControllerService() {
ClusterControllerService ccs = mock(ClusterControllerService.class);
CCServiceContext ccServiceCtx = mock(CCServiceContext.class);