Fixed update of a migrated async external worker job with the new pro…

…cess definition id (flowable#3894)
basclaessen · May 19, 2024 · d43fb00 · d43fb00
1 parent 7683479
commit d43fb00
Show file tree

Hide file tree

Showing 3 changed files with 101 additions and 10 deletions.
diff --git a/...le-engine/src/main/java/org/flowable/engine/impl/dynamic/AbstractDynamicStateManager.java b/...le-engine/src/main/java/org/flowable/engine/impl/dynamic/AbstractDynamicStateManager.java
@@ -1168,6 +1168,14 @@ protected void handleServiceTaskJobUpdate(ExecutionEntity childExecution, Comman
             suspendedJob.setProcessDefinitionId(childExecution.getProcessDefinitionId());
             return;
         }
+
+        // Update an existing external worker job
+        ExternalWorkerJobEntityImpl externalWorkerJob = (ExternalWorkerJobEntityImpl) managementService.createExternalWorkerJobQuery()
+                .executionId(childExecution.getId()).singleResult();
+        if (externalWorkerJob != null) {
+            externalWorkerJob.setProcessDefinitionId(childExecution.getProcessDefinitionId());
+            return;
+        }
     }
 
     protected void handleExternalWorkerServiceTaskJobUpdate(ExecutionEntity childExecution, CommandContext commandContext) {

diff --git a/...test/api/runtime/migration/ProcessInstanceDirectExecutionMigrationExternalWorkerTest.java b/...test/api/runtime/migration/ProcessInstanceDirectExecutionMigrationExternalWorkerTest.java
@@ -54,16 +54,25 @@ protected void tearDown() {
 
     @Test
     public void testSimpleMigrationWithActivityAutoMapping() {
+        testSimpleMigrationWithActivityAutoMapping(false, "org/flowable/engine/test/api/runtime/migration/one-external-worker-simple-process.bpmn20.xml");
+    }
+
+    @Test
+    public void testSimpleMigrationWithActivityAutoMappingAsync() {
+        testSimpleMigrationWithActivityAutoMapping(true, "org/flowable/engine/test/api/runtime/migration/one-async-external-worker-simple-process.bpmn20.xml");
+    }
+
+    protected void testSimpleMigrationWithActivityAutoMapping(boolean async, String processDefinitionPath) {
         //Deploy first version of the process
         ProcessDefinition version1ProcessDef = deployProcessDefinition("my deploy",
-                "org/flowable/engine/test/api/runtime/migration/one-external-worker-simple-process.bpmn20.xml");
+                processDefinitionPath);
 
         //Start an instance of the first version of the process for migration
         ProcessInstance processInstance = runtimeService.startProcessInstanceByKey("MP");
 
         //Deploy second version of the same process
         ProcessDefinition version2ProcessDef = deployProcessDefinition("my deploy",
-                "org/flowable/engine/test/api/runtime/migration/one-external-worker-simple-process.bpmn20.xml");
+                processDefinitionPath);
         assertThat(version1ProcessDef.getId()).isNotEqualTo(version2ProcessDef.getId());
 
         List<ProcessDefinition> processDefinitions = repositoryService.createProcessDefinitionQuery()
@@ -79,6 +88,11 @@ public void testSimpleMigrationWithActivityAutoMapping() {
                 .map(e -> (ExecutionEntity) e)
                 .forEach(e -> assertThat(e.getProcessDefinitionId()).isEqualTo(version1ProcessDef.getId()));
 
+        if (async) {
+            // Start the async executor to create the external worker job
+            JobTestHelper.waitForJobExecutorToProcessAllJobs(processEngineConfiguration, managementService, 1000L, 500L, true);
+        }
+
         List<ExternalWorkerJob> externalWorkerJobs = managementService.createExternalWorkerJobQuery()
                 .list();
         assertThat(externalWorkerJobs)
@@ -131,16 +145,25 @@ public void testSimpleMigrationWithActivityAutoMapping() {
 
     @Test
     public void testAcquiredBeforeMigration() {
+        testAcquiredBeforeMigration(false, "org/flowable/engine/test/api/runtime/migration/one-external-worker-simple-process.bpmn20.xml");
+    }
+
+    @Test
+    public void testAcquiredBeforeMigrationAsync() {
+        testAcquiredBeforeMigration(true, "org/flowable/engine/test/api/runtime/migration/one-async-external-worker-simple-process.bpmn20.xml");
+    }
+
+    protected void testAcquiredBeforeMigration(boolean async, String processDefinitionPath) {
         //Deploy first version of the process
         ProcessDefinition version1ProcessDef = deployProcessDefinition("my deploy",
-                "org/flowable/engine/test/api/runtime/migration/one-external-worker-simple-process.bpmn20.xml");
+                processDefinitionPath);
 
         //Start an instance of the first version of the process for migration
         ProcessInstance processInstance = runtimeService.startProcessInstanceByKey("MP");
 
         //Deploy second version of the same process
         ProcessDefinition version2ProcessDef = deployProcessDefinition("my deploy",
-                "org/flowable/engine/test/api/runtime/migration/one-external-worker-simple-process.bpmn20.xml");
+                processDefinitionPath);
         assertThat(version1ProcessDef.getId()).isNotEqualTo(version2ProcessDef.getId());
 
         List<ProcessDefinition> processDefinitions = repositoryService.createProcessDefinitionQuery()
@@ -156,6 +179,11 @@ public void testAcquiredBeforeMigration() {
                 .map(e -> (ExecutionEntity) e)
                 .forEach(e -> assertThat(e.getProcessDefinitionId()).isEqualTo(version1ProcessDef.getId()));
 
+        if (async) {
+            // Start the async executor to create the external worker job
+            JobTestHelper.waitForJobExecutorToProcessAllJobs(processEngineConfiguration, managementService, 1000L, 500L, true);
+        }
+
         List<ExternalWorkerJob> externalWorkerJobs = managementService.createExternalWorkerJobQuery()
                 .list();
         assertThat(externalWorkerJobs)
@@ -208,16 +236,25 @@ public void testAcquiredBeforeMigration() {
 
     @Test
     public void testSuspendedBeforeMigration() {
+        testSuspendedBeforeMigration(false, "org/flowable/engine/test/api/runtime/migration/one-external-worker-simple-process.bpmn20.xml");
+    }
+
+    @Test
+    public void testSuspendedBeforeMigrationAsync() {
+        testSuspendedBeforeMigration(true, "org/flowable/engine/test/api/runtime/migration/one-async-external-worker-simple-process.bpmn20.xml");
+    }
+
+    protected void testSuspendedBeforeMigration(boolean async, String processDefinitionPath) {
         //Deploy first version of the process
         ProcessDefinition version1ProcessDef = deployProcessDefinition("my deploy",
-                "org/flowable/engine/test/api/runtime/migration/one-external-worker-simple-process.bpmn20.xml");
+                processDefinitionPath);
 
         //Start an instance of the first version of the process for migration
         ProcessInstance processInstance = runtimeService.startProcessInstanceByKey("MP");
 
         //Deploy second version of the same process
         ProcessDefinition version2ProcessDef = deployProcessDefinition("my deploy",
-                "org/flowable/engine/test/api/runtime/migration/one-external-worker-simple-process.bpmn20.xml");
+                processDefinitionPath);
         assertThat(version1ProcessDef.getId()).isNotEqualTo(version2ProcessDef.getId());
 
         List<ProcessDefinition> processDefinitions = repositoryService.createProcessDefinitionQuery()
@@ -233,6 +270,11 @@ public void testSuspendedBeforeMigration() {
                 .map(e -> (ExecutionEntity) e)
                 .forEach(e -> assertThat(e.getProcessDefinitionId()).isEqualTo(version1ProcessDef.getId()));
 
+        if (async) {
+            // Start the async executor to create the external worker job
+            JobTestHelper.waitForJobExecutorToProcessAllJobs(processEngineConfiguration, managementService, 1000L, 500L, true);
+        }
+
         List<ExternalWorkerJob> externalWorkerJobs = managementService.createExternalWorkerJobQuery()
                 .list();
         assertThat(externalWorkerJobs)
@@ -291,18 +333,32 @@ public void testSuspendedBeforeMigration() {
 
     @Test
     public void testDeadLetterBeforeMigration() {
+        testDeadLetterBeforeMigration(false, "org/flowable/engine/test/api/runtime/migration/one-external-worker-simple-process.bpmn20.xml");
+    }
+
+    @Test
+    public void testDeadLetterBeforeMigrationAsync() {
+        testDeadLetterBeforeMigration(true, "org/flowable/engine/test/api/runtime/migration/one-async-external-worker-simple-process.bpmn20.xml");
+    }
+
+    protected void testDeadLetterBeforeMigration(boolean async, String processDefinitionPath) {
         //Deploy first version of the process
         ProcessDefinition version1ProcessDef = deployProcessDefinition("my deploy",
-                "org/flowable/engine/test/api/runtime/migration/one-external-worker-simple-process.bpmn20.xml");
+                processDefinitionPath);
 
         //Start an instance of the first version of the process for migration
         ProcessInstance processInstance = runtimeService.startProcessInstanceByKey("MP");
 
         //Deploy second version of the same process
         ProcessDefinition version2ProcessDef = deployProcessDefinition("my deploy",
-                "org/flowable/engine/test/api/runtime/migration/one-external-worker-simple-process.bpmn20.xml");
+                processDefinitionPath);
         assertThat(version1ProcessDef.getId()).isNotEqualTo(version2ProcessDef.getId());
 
+        if (async) {
+            // Start the async executor to create the external worker job
+            JobTestHelper.waitForJobExecutorToProcessAllJobs(processEngineConfiguration, managementService, 1000L, 500L, true);
+        }
+
         List<ExternalWorkerJob> externalWorkerJobs = managementService.createExternalWorkerJobQuery()
                 .list();
         assertThat(externalWorkerJobs)
@@ -370,18 +426,32 @@ public void testDeadLetterBeforeMigration() {
 
     @Test
     public void testRetryBeforeMigration() {
+        testRetryBeforeMigration(false, "org/flowable/engine/test/api/runtime/migration/one-external-worker-simple-process.bpmn20.xml");
+    }
+
+    @Test
+    public void testRetryBeforeMigrationAsync() {
+        testRetryBeforeMigration(true, "org/flowable/engine/test/api/runtime/migration/one-async-external-worker-simple-process.bpmn20.xml");
+    }
+
+    protected void testRetryBeforeMigration(boolean async, String processDefinitionPath) {
         //Deploy first version of the process
         ProcessDefinition version1ProcessDef = deployProcessDefinition("my deploy",
-                "org/flowable/engine/test/api/runtime/migration/one-external-worker-simple-process.bpmn20.xml");
+                processDefinitionPath);
 
         //Start an instance of the first version of the process for migration
         ProcessInstance processInstance = runtimeService.startProcessInstanceByKey("MP");
 
         //Deploy second version of the same process
         ProcessDefinition version2ProcessDef = deployProcessDefinition("my deploy",
-                "org/flowable/engine/test/api/runtime/migration/one-external-worker-simple-process.bpmn20.xml");
+                processDefinitionPath);
         assertThat(version1ProcessDef.getId()).isNotEqualTo(version2ProcessDef.getId());
 
+        if (async) {
+            // Start the async executor to create the external worker job
+            JobTestHelper.waitForJobExecutorToProcessAllJobs(processEngineConfiguration, managementService, 1000L, 500L, true);
+        }
+
         List<ExternalWorkerJob> externalWorkerJobs = managementService.createExternalWorkerJobQuery()
                 .list();
         assertThat(externalWorkerJobs)

diff --git a/...ble/engine/test/api/runtime/migration/one-async-external-worker-simple-process.bpmn20.xml b/...ble/engine/test/api/runtime/migration/one-async-external-worker-simple-process.bpmn20.xml
@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<definitions id="definitions" xmlns="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:flowable="http://flowable.org/bpmn" targetNamespace="Examples">
+
+    <process id="MP" name="MyProcess" isExecutable="true">
+        <documentation>My process documentation</documentation>
+        <startEvent id="startEvent1"/>
+        <sequenceFlow id="seqFlow1Id" sourceRef="startEvent1" targetRef="externalWorker1Id"/>
+        <serviceTask id="externalWorker1Id" name="externalWorkerName" flowable:type="external-worker" flowable:topic="topic1" flowable:async="true"/>
+        <sequenceFlow id="seqFlow2Id" sourceRef="externalWorker1Id" targetRef="endEvent1"/>
+        <endEvent id="endEvent1"/>
+    </process>
+
+</definitions>