Support job_retries_decremented event for CMMN

Fixes #4062

Author: martin-grofcik

modules/flowable-cmmn-engine/src/main/java/org/flowable/cmmn/engine/impl/cmd/JobRetryCmd.java

@@ -18,19 +18,24 @@
 import java.util.Date;
 import java.util.GregorianCalendar;
 
+import org.flowable.cmmn.engine.CmmnEngineConfiguration;
 import org.flowable.cmmn.engine.impl.util.CommandContextUtil;
+import org.flowable.common.engine.api.delegate.event.FlowableEngineEventType;
+import org.flowable.common.engine.api.delegate.event.FlowableEventDispatcher;
 import org.flowable.common.engine.impl.interceptor.Command;
 import org.flowable.common.engine.impl.interceptor.CommandContext;
 import org.flowable.common.engine.impl.util.ExceptionUtil;
 import org.flowable.job.api.FlowableUnrecoverableJobException;
 import org.flowable.job.service.JobService;
 import org.flowable.job.service.TimerJobService;
+import org.flowable.job.service.event.impl.FlowableJobEventBuilder;
 import org.flowable.job.service.impl.persistence.entity.AbstractRuntimeJobEntity;
 import org.flowable.job.service.impl.persistence.entity.JobEntity;
 
 /**
  * @author Saeid Mirzaei
  * @author Joram Barrez
+ * @author martin.grofcik
  */
 public class JobRetryCmd implements Command<Object> {
 
@@ -52,7 +57,7 @@ public Object execute(CommandContext commandContext) {
             return null;
         }
 
-        AbstractRuntimeJobEntity newJobEntity = null;
+        AbstractRuntimeJobEntity newJobEntity;
         if (job.getRetries() <= 1 || isUnrecoverableException()) {
             newJobEntity = jobService.moveJobToDeadLetterJob(job);
         } else {
@@ -73,6 +78,16 @@ public Object execute(CommandContext commandContext) {
             newJobEntity.setExceptionStacktrace(getExceptionStacktrace());
         }
 
+        // Dispatch both an update and a retry-decrement event
+        CmmnEngineConfiguration configuration = CommandContextUtil.getCmmnEngineConfiguration(commandContext);
+        FlowableEventDispatcher eventDispatcher = configuration.getEventDispatcher();
+        if (eventDispatcher != null && eventDispatcher.isEnabled()) {
+            eventDispatcher.dispatchEvent(FlowableJobEventBuilder.createEntityEvent(FlowableEngineEventType.ENTITY_UPDATED, newJobEntity),
+                    configuration.getEngineCfgKey());
+            eventDispatcher.dispatchEvent(FlowableJobEventBuilder.createEntityEvent(FlowableEngineEventType.JOB_RETRIES_DECREMENTED, newJobEntity),
+                    configuration.getEngineCfgKey());
+        }
+
         return null;
     }
 

modules/flowable-cmmn-engine/src/test/java/org/flowable/cmmn/test/event/JobRetriesDecrementedEventTest.java

@@ -0,0 +1,111 @@
+/* Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.flowable.cmmn.test.event;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.flowable.common.engine.api.delegate.event.FlowableEngineEventType.ENTITY_UPDATED;
+import static org.flowable.common.engine.api.delegate.event.FlowableEngineEventType.JOB_RETRIES_DECREMENTED;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+
+import org.flowable.cmmn.api.runtime.CaseInstance;
+import org.flowable.cmmn.engine.test.CmmnDeployment;
+import org.flowable.cmmn.test.FlowableCmmnTestCase;
+import org.flowable.common.engine.api.delegate.event.AbstractFlowableEventListener;
+import org.flowable.common.engine.api.delegate.event.FlowableEngineEntityEvent;
+import org.flowable.common.engine.api.delegate.event.FlowableEvent;
+import org.flowable.common.engine.api.delegate.event.FlowableEventType;
+import org.flowable.job.api.Job;
+import org.flowable.task.api.Task;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * @author martin.grofcik
+ */
+public class JobRetriesDecrementedEventTest extends FlowableCmmnTestCase {
+
+    TestEventListener listener = new TestEventListener();
+
+    @BeforeEach
+    void addListener() {
+        listener.events.clear();
+        cmmnEngineConfiguration.getEventDispatcher().addEventListener(listener);
+    }
+
+    @AfterEach
+    void removeListener() {
+        cmmnEngineConfiguration.getEventDispatcher().removeEventListener(listener);
+    }
+
+    @Test
+    @CmmnDeployment(resources = "org/flowable/cmmn/test/async/AsyncTaskTest.testAsyncServiceTask.cmmn")
+    void noEventForSuccess() {
+        CaseInstance caseInstance = cmmnRuntimeService.createCaseInstanceBuilder().caseDefinitionKey("testAsyncServiceTask").start();
+        Task task = cmmnTaskService.createTaskQuery()
+                .caseInstanceId(caseInstance.getId())
+                .singleResult();
+        assertThat(task.getName()).isEqualTo("Task before service task");
+        cmmnTaskService.complete(task.getId());
+
+        waitForJobExecutorToProcessAllAsyncJobs();
+
+        assertThat(listener.events).isEmpty();
+    }
+
+    @Test
+    @CmmnDeployment(resources = "org/flowable/cmmn/test/async/AsyncTaskTest.testAsyncServiceTaskWithFailure.cmmn")
+    void jobRetriesDecrementedOnFailure() {
+        CaseInstance caseInstance = cmmnRuntimeService.createCaseInstanceBuilder().caseDefinitionKey("testAsyncServiceTask").start();
+        Task task = cmmnTaskService.createTaskQuery()
+                .caseInstanceId(caseInstance.getId())
+                .singleResult();
+        assertThat(task.getName()).isEqualTo("Task before service task");
+        cmmnTaskService.complete(task.getId());
+
+        waitForJobExecutorToProcessAllAsyncJobs();
+
+        assertThat(listener.events)
+                .filteredOn( e -> caseInstance.getId().equals(getScopeId(e)))
+                .extracting(FlowableEvent::getType)
+                .containsOnly(JOB_RETRIES_DECREMENTED, ENTITY_UPDATED);
+    }
+
+    private String getScopeId(FlowableEvent e) {
+        return ((Job) ((FlowableEngineEntityEvent) e).getEntity()).getScopeId();
+    }
+
+    private static class TestEventListener extends AbstractFlowableEventListener {
+        private static final Collection<? extends FlowableEventType> SUPPORTED_TYPES = Set.of(JOB_RETRIES_DECREMENTED, ENTITY_UPDATED);
+        List<FlowableEvent> events = new ArrayList<>();
+
+        @Override
+        public void onEvent(FlowableEvent event) {
+            events.add(event);
+        }
+
+        @Override
+        public boolean isFailOnException() {
+            return false;
+        }
+
+        @Override
+        public Collection<? extends FlowableEventType> getTypes() {
+            return SUPPORTED_TYPES;
+        }
+    }
+}