Unexpected behavior when retrying a CallActivity depending on the asyncComplete-flag

Comienza con los dos recursos BPMN de CallActivityTest y los métodos de prueba mostrados en el issue; después reproduce ambos casos de completeAsync usando la configuración de retry proporcionada. Lee JobRetryCmd y ParallelMultiInstanceBehavior, especialmente internalLeave, para seguir la actividad de ejecución y el recuento de instancias completadas. La tarea estará terminada cuando la prueba de retry pase tanto para la finalización asíncrona como para la síncrona, sin cancelar los subflujos restantes.

I am observing different results when retrying a CallActivity with multi-instance behavior, depending on the value of completeAsync.

This issue is similar to this other topic. However, they are focusing on inconsistencies of the event context in case of a retry of a failed CallActivity. I decided to create a dedicated issue for this because they only mention my problem in a comment in their linked PR.

Describe the bug
If the CallActivity (with multi-instance behavior) is configured with completeAsync=true and retried after an error, only one subflow completes successfully, the other subflows get canceled

Unit Test:
Here is a commit with a complete test setup. For completeness of this post, I will outline the scenario here as well.

The parent process contains a CallActivity with multi-instance behavior, which starts 3 subflows. The CallActivity is either configured with completeAsync=true or completeAsync=false. Additionally, the CallActivity contains a VariableAggregation, which sole purpose is to provoke an error inside ParallelMultiInstanceBehavior::leave. The subflow only contains a start and end event.

The test with completeAsync=true fails (only one of the subflows completes successfully, the other two subflows get canceled), but the test with completeAsync=false passes.

    @Test
    @Deployment(resources = {
            "org/flowable/engine/test/api/event/CallActivityTest.testCallActivityWithMultiInstanceBehaviorAndAsyncCompleteRetry.bpmn20.xml",
            "org/flowable/engine/test/api/event/CallActivityTest.testCallActivityWithMultiInstanceBehaviorAndAsyncCompleteRetry_subflow.bpmn20.xml",
    })
    public void testCallActivityWithMultiInstanceBehaviorAndAsyncCompleteRetry() throws Exception {
        testCallActivityWithMultiInstanceBehaviorRetryAfterFailure();
    }
    
    @Test
    @Deployment(resources = {
            "org/flowable/engine/test/api/event/CallActivityTest.testCallActivityWithMultiInstanceBehaviorAndSyncCompleteRetry.bpmn20.xml",
            "org/flowable/engine/test/api/event/CallActivityTest.testCallActivityWithMultiInstanceBehaviorAndAsyncCompleteRetry_subflow.bpmn20.xml",
    })
    public void testCallActivityWithMultiInstanceBehaviorAndSyncCompleteRetry() throws Exception {
        testCallActivityWithMultiInstanceBehaviorRetryAfterFailure();
    }
    
    public void testCallActivityWithMultiInstanceBehaviorRetryAfterFailure() throws Exception {
        // Set number of retries to 1. In case of an error, the Job becomes a DeadLetterJob
        processEngineConfiguration.getJobServiceConfiguration().setAsyncExecutorNumberOfRetries(1);

        ProcessInstance processInstance = runtimeService.startProcessInstanceByKey("callActivity");

        waitForJobExecutorToProcessAllJobsAndExecutableTimerJobs(20000L, 200L);

        // Check erroneous state:
        // The CallActivity is configured with VariableAggregation. However, it is intentionally referencing a non-existing variable. This will cause the CallActivity to fail.
        assertThat(processEngine.getManagementService().createDeadLetterJobQuery().withException().count() == 3).isTrue();

        // Set the missing process variable. After it is set, the process should complete successfully.
        runtimeService.setVariable(processInstance.getId(), "nonExistingVariable", "target");

        // Retry failed jobs
        managementService.createDeadLetterJobQuery().list().forEach(job -> managementService.moveDeadLetterJobToExecutableJob(job.getId(), 3));
        waitForJobExecutorToProcessAllJobsAndExecutableTimerJobs(20000L, 200L);
        
        // Check that the parent process and its subflows completed
        assertProcessEnded(processInstance.getId());
    }

Expected behavior
I expect that my test passes when completeAsync=true and when completeAsync=false.

Error Analysis: I will now explain what goes wrong, if completeAsync is true:

When the exception is thrown due to the VariableAggregation, the execution of the current job is set to “inactive” (JobRetryCmd::execute). In this test, the execution of the current job refers to the “CallActivity” (in the attached image, these are the executions with the IDs 17, 18, and 19). This differs from the test where completeAsync is false (there, the execution of the jobs refer to the EndEvent of the subflows (see IDs 23, 29, and 35 in the attached image)). After the retry (i.e. moving the DeadLetterJobs), all three CallActivity-executions are executed again. However, they remain “inactive” (i.e. the column IS_ACTIVE in the DB has the value false).

Every time an instance of the CallActivity finishes, the number of “completed” instances is compared to the total number of instances (this happens in ParallelMultiInstanceBehavior::internalLeave). If nrOfCompletedInstances >= nrOfInstances || isCompletionConditionSatisfied evaluates to true all remaining instances of the CallActivity get cancelled. However, here is the problem: when the first instance reaches this point (after the retry), the value of nrOfCompletedInstances is 4 (which is wrong) and the value of nrOfInstances is 3 (which is correct). Hence, the other two instances get cancelled.
The value of nrOfCompletedInstances is calculated by counting the number of CallActivity-executions where IS_ACTIVE==false (i.e. 17, 18, 19) (+ 1 for the current instance).

Note:
I originally posted this on the Flowable Forum. Since I did not receive any feedback there, I guess that this is a better place to raise the problem.

Additional context