A non transactional exclusive job should be unlocked after the execution is done and in the same transaction as the deletion

Author: filiphr

modules/flowable-engine/src/test/java/org/flowable/engine/test/jobexecutor/NonTransactionalJobHandlerTest.java

@@ -14,18 +14,24 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.util.Date;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.flowable.common.engine.impl.cfg.TransactionContext;
 import org.flowable.common.engine.impl.interceptor.CommandContext;
 import org.flowable.engine.impl.context.Context;
+import org.flowable.engine.impl.persistence.entity.ExecutionEntity;
 import org.flowable.engine.impl.test.JobTestHelper;
 import org.flowable.engine.impl.test.PluggableFlowableTestCase;
 import org.flowable.engine.impl.util.CommandContextUtil;
+import org.flowable.engine.runtime.ProcessInstance;
+import org.flowable.engine.test.Deployment;
 import org.flowable.job.api.Job;
 import org.flowable.job.service.JobService;
 import org.flowable.job.service.impl.nontx.NonTransactionalJobHandler;
 import org.flowable.job.service.impl.persistence.entity.JobEntity;
+import org.flowable.task.api.Task;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -83,6 +89,54 @@ public void testJobExecutedWithoutTransaction() {
         assertThat(nonTransactionalTestJobHandler.nonTransactionalCounter).hasValue(1);
     }
 
+    @Test
+    @Deployment(resources = "org/flowable/engine/test/api/oneTaskProcess.bpmn20.xml")
+    public void testAsyncExclusiveJob() {
+        ProcessInstance processInstance = runtimeService.createProcessInstanceBuilder()
+                .processDefinitionKey("oneTaskProcess")
+                .start();
+        Task task = taskService.createTaskQuery()
+                .processInstanceId(processInstance.getId())
+                .singleResult();
+        managementService.executeCommand(commandContext -> {
+            JobService jobService = CommandContextUtil.getJobService();
+            JobEntity job = jobService.createJob();
+            job.setProcessInstanceId(processInstance.getId());
+            job.setExecutionId(task.getExecutionId());
+            job.setJobHandlerType(nonTransactionalTestJobHandler.getType());
+            job.setJobHandlerConfiguration("myTest");
+            jobService.createAsyncJob(job, true);
+            jobService.scheduleAsyncJob(job);
+            return null;
+        });
+
+        AtomicReference<String> processLockOwner = new AtomicReference<>();
+        AtomicReference<Date> processLockTime = new AtomicReference<>();
+        nonTransactionalTestJobHandler.nonTransactionalRunnable = () -> {
+            ExecutionEntity executionEntity = (ExecutionEntity) runtimeService.createProcessInstanceQuery()
+                    .processInstanceId(processInstance.getId())
+                    .singleResult();
+            processLockOwner.set(executionEntity.getLockOwner());
+            processLockTime.set(executionEntity.getLockTime());
+        };
+
+        JobTestHelper.waitForJobExecutorOnCondition(processEngineConfiguration, 10000L, 20L,
+                () -> managementService.createJobQuery().count() == 0);
+
+        assertThat(processLockOwner).hasValue(processEngineConfiguration.getAsyncExecutor().getLockOwner());
+        assertThat(processLockTime).doesNotHaveNullValue();
+
+        managementService.executeCommand(commandContext -> {
+            ExecutionEntity executionEntity = (ExecutionEntity) runtimeService.createProcessInstanceQuery()
+                    .processInstanceId(processInstance.getId())
+                    .singleResult();
+
+            assertThat(executionEntity.getLockOwner()).isNull();
+            assertThat(executionEntity.getLockTime()).isNull();
+            return null;
+        });
+    }
+
     @Test
     public void testJobExecutedWithoutTransactionThrowsException() {
         managementService.executeCommand(commandContext -> {
@@ -125,6 +179,8 @@ public static class NonTransactionalTestJobHandler implements NonTransactionalJo
         protected AtomicInteger withoutTransactionCounter = new AtomicInteger(0);
         protected AtomicInteger nonTransactionalCounter = new AtomicInteger(0);
 
+        protected Runnable nonTransactionalRunnable;
+
         protected String jobConfiguration;
         protected String nonTransactionalOutput;
 
@@ -154,6 +210,9 @@ public String executeNonTransactionally(JobEntity job, String configuration) {
             }
 
             this.jobConfiguration = job.getJobHandlerConfiguration();
+            if (nonTransactionalRunnable != null) {
+                nonTransactionalRunnable.run();
+            }
             return jobConfiguration;
 
         }
@@ -189,6 +248,7 @@ public String getJobConfiguration() {
         }
 
         public void reset() {
+            this.nonTransactionalRunnable = null;
             this.counter.set(0);
             this.withCommandContext.set(0);
             this.withoutCommandContext.set(0);

modules/flowable-job-service/src/main/java/org/flowable/job/service/impl/asyncexecutor/ExecuteAsyncRunnable.java

@@ -182,10 +182,6 @@ protected boolean handleNontransactionalJob(NonTransactionalJobHandler<Object> j
 
         JobProcessorUtil.callJobProcessors(jobServiceConfiguration, JobProcessorContext.Phase.BEFORE_EXECUTE, (JobEntity) job);
 
-        if (unlock) {
-            jobServiceConfiguration.getCommandExecutor().execute(new UnlockExclusiveJobCmd((Job) job, jobServiceConfiguration));
-        }
-
         // If an exception is thrown during job handler exception, it goes up and will be caught in the general exception handling.
         // The delete at the end won't happen in that case.
 
@@ -200,6 +196,14 @@ protected boolean handleNontransactionalJob(NonTransactionalJobHandler<Object> j
         // The delete still needs to happen in a new transaction
         jobServiceConfiguration.getCommandExecutor().execute(commandContext -> {
             jobHandler.afterExecute((JobEntity) job, job.getJobHandlerConfiguration(), nonTransactionalOutput, commandContext);
+
+            if (unlock) {
+                // Part of the same transaction to avoid a race condition with the
+                // potentially new jobs (wrt process instance locking) that are created
+                // during the execution of the original job
+                new UnlockExclusiveJobCmd((Job) job, jobServiceConfiguration).execute(commandContext);
+            }
+
             jobServiceConfiguration.getJobEntityManager().delete((JobEntity) job);
             return null;
         });