merge: #8460

8460: Support inside properties for multi-instance completion condition r=korthout a=Lzgabel

## Description

The BPMN 2.0 specification defined the following properties of a multi-instance body instance:

* numberOfInstances
* numberOfActiveInstances
* numberOfCompletedInstances
* numberOfTerminatedInstances

This PR makes `numberOfInstances` and `numberOfActiveInstances` available for use in the [completion condition](https://docs.camunda.io/docs/components/modeler/bpmn/multi-instance/#completion-condition) expression. Although they are available in that expression, they do not exist as process variables. These properties take precedence over process variables with the same name.

Out of scope: `numberOfCompletedInstances`, `numberOfTerminatedInstances`
Out of scope: using these properties in other expressions

## Related issues

closes #2893

## Definition of Done

_Not all items need to be done depending on the issue and the pull request._

Code changes:
* [x] The changes are backwards compatibility with previous versions
* [ ] If it fixes a bug then PRs are created to [backport](https://github.com/zeebe-io/zeebe/compare/stable/0.24...develop?expand=1&template=backport_template.md&title=[Backport%200.24]) the fix to the last two minor versions. You can trigger a backport by assigning labels (e.g. `backport stable/0.25`) to the PR, in case that fails you need to create backports manually.

Testing:
* [x] There are unit/integration tests that verify all acceptance criterias of the issue
* [x] New tests are written to ensure backwards compatibility with further versions
* [x] The behavior is tested manually
* [x] The change has been verified by a QA run
* [ ] The impact of the changes is verified by a benchmark 

Documentation: 
* [ ] The documentation is updated (e.g. BPMN reference, configuration, examples, get-started guides, etc.)
* [x] New content is added to the [release announcement](https://drive.google.com/drive/u/0/folders/1DTIeswnEEq-NggJ25rm2BsDjcCQpDape)


Co-authored-by: lzgabel <lz19960321lz@gmail.com>

engine/src/main/java/io/camunda/zeebe/engine/processing/bpmn/container/MultiInstanceBodyProcessor.java

@@ -39,9 +39,16 @@ public final class MultiInstanceBodyProcessor
 
   private final MutableDirectBuffer loopCounterVariableBuffer =
       new UnsafeBuffer(new byte[Long.BYTES + 1]);
+  private final MutableDirectBuffer numberOfInstancesVariableBuffer =
+      new UnsafeBuffer(new byte[Long.BYTES + 1]);
+  private final MutableDirectBuffer numberOfActiveInstancesVariableBuffer =
+      new UnsafeBuffer(new byte[Long.BYTES + 1]);
+
   private final DirectBuffer loopCounterVariableView = new UnsafeBuffer(0, 0);
+  private final DirectBuffer numberOfInstancesVariableView = new UnsafeBuffer(0, 0);
+  private final DirectBuffer numberOfActiveInstancesVariableView = new UnsafeBuffer(0, 0);
 
-  private final MsgPackWriter loopCounterWriter = new MsgPackWriter();
+  private final MsgPackWriter variableWriter = new MsgPackWriter();
 
   private final ExpressionProcessor expressionBehavior;
   private final BpmnStateTransitionBehavior stateTransitionBehavior;
@@ -312,7 +319,9 @@ private void setLoopVariables(
             variableName -> stateBehavior.setLocalVariable(childContext, variableName, NIL_VALUE));
 
     stateBehavior.setLocalVariable(
-        childContext, LOOP_COUNTER_VARIABLE, wrapLoopCounter(loopCounter));
+        childContext,
+        LOOP_COUNTER_VARIABLE,
+        wrapVariable(loopCounterVariableBuffer, loopCounterVariableView, loopCounter));
   }
 
   private Either<Failure, List<DirectBuffer>> readInputCollectionVariable(
@@ -328,25 +337,59 @@ private void createInnerInstance(
         context, multiInstanceBody.getInnerActivity());
   }
 
-  private DirectBuffer wrapLoopCounter(final int loopCounter) {
-    loopCounterWriter.wrap(loopCounterVariableBuffer, 0);
+  private DirectBuffer wrapVariable(
+      final MutableDirectBuffer variableBuffer, final DirectBuffer variableView, final long value) {
+    variableWriter.wrap(variableBuffer, 0);
 
-    loopCounterWriter.writeInteger(loopCounter);
-    final var length = loopCounterWriter.getOffset();
+    variableWriter.writeInteger(value);
+    final var length = variableWriter.getOffset();
 
-    loopCounterVariableView.wrap(loopCounterVariableBuffer, 0, length);
-    return loopCounterVariableView;
+    variableView.wrap(variableBuffer, 0, length);
+    return variableView;
   }
 
   private Either<Failure, Boolean> satisfiesCompletionCondition(
       final ExecutableMultiInstanceBody element, final BpmnElementContext context) {
     final Optional<Expression> completionCondition =
         element.getLoopCharacteristics().getCompletionCondition();
 
+    final ExpressionProcessor primaryContextExpressionProcessor =
+        expressionBehavior.withPrimaryContext(
+            (variableName -> getVariable(context.getFlowScopeKey(), variableName)));
     if (completionCondition.isPresent()) {
-      return expressionBehavior.evaluateBooleanExpression(
+      return primaryContextExpressionProcessor.evaluateBooleanExpression(
           completionCondition.get(), context.getElementInstanceKey());
     }
     return Either.right(false);
   }
+
+  private DirectBuffer getVariable(final long elementInstanceKey, final String variableName) {
+    return switch (variableName) {
+      case "numberOfInstances" -> getNumberOfInstancesVariable(elementInstanceKey);
+
+      case "numberOfActiveInstances" -> getNumberOfActiveInstancesVariable(elementInstanceKey);
+
+      default -> null;
+    };
+  }
+
+  private DirectBuffer getNumberOfInstancesVariable(final long elementInstanceKey) {
+    return wrapVariable(
+        numberOfInstancesVariableBuffer,
+        numberOfInstancesVariableView,
+        stateBehavior.getElementInstance(elementInstanceKey).getNumberOfElementInstances());
+  }
+
+  private DirectBuffer getNumberOfActiveInstancesVariable(final long elementInstanceKey) {
+    // The getNumberOfActiveInstancesVariable method is called while the child instance is
+    // completing, but the active element instances value has not yet been decremented,
+    // which is why this variable has to be lowered by 1
+    final int numberOfActiveInstances =
+        stateBehavior.getElementInstance(elementInstanceKey).getNumberOfActiveElementInstances()
+            - 1;
+    return wrapVariable(
+        numberOfActiveInstancesVariableBuffer,
+        numberOfActiveInstancesVariableView,
+        numberOfActiveInstances);
+  }
 }

engine/src/main/java/io/camunda/zeebe/engine/state/appliers/ProcessInstanceElementActivatingApplier.java

@@ -71,7 +71,7 @@ public void applyState(final long elementInstanceKey, final ProcessInstanceRecor
           flowScopeEventTrigger, flowScopeInstance.getParentKey(), elementInstanceKey);
     }
 
-    manageMultiInstanceLoopCounter(elementInstanceKey, flowScopeInstance, flowScopeElementType);
+    manageMultiInstance(elementInstanceKey, flowScopeInstance, flowScopeElementType);
   }
 
   private void cleanupSequenceFlowsTaken(final ProcessInstanceRecord value) {
@@ -206,13 +206,15 @@ private ExecutableFlowElementContainer getExecutableFlowElementContainer(
         ExecutableFlowElementContainer.class);
   }
 
-  private void manageMultiInstanceLoopCounter(
+  private void manageMultiInstance(
       final long elementInstanceKey,
       final ElementInstance flowScopeInstance,
       final BpmnElementType flowScopeElementType) {
     if (flowScopeElementType == BpmnElementType.MULTI_INSTANCE_BODY) {
       // update the loop counter of the multi-instance body (starting by 1)
       flowScopeInstance.incrementMultiInstanceLoopCounter();
+      // update the numberOfInstances of the multi-instance body
+      flowScopeInstance.incrementNumberOfElementInstances();
       elementInstanceState.updateInstance(flowScopeInstance);
 
       // set the loop counter of the inner instance

engine/src/main/java/io/camunda/zeebe/engine/state/instance/ElementInstance.java

@@ -22,6 +22,8 @@ public final class ElementInstance extends UnpackedObject implements DbValue {
 
   private final LongProperty parentKeyProp = new LongProperty("parentKey", -1L);
   private final IntegerProperty childCountProp = new IntegerProperty("childCount", 0);
+  private final IntegerProperty childActivatedCountProp =
+      new IntegerProperty("childActivatedCount", 0);
   private final LongProperty jobKeyProp = new LongProperty("jobKey", 0L);
   private final IntegerProperty multiInstanceLoopCounterProp =
       new IntegerProperty("multiInstanceLoopCounter", 0);
@@ -37,6 +39,7 @@ public final class ElementInstance extends UnpackedObject implements DbValue {
   ElementInstance() {
     declareProperty(parentKeyProp)
         .declareProperty(childCountProp)
+        .declareProperty(childActivatedCountProp)
         .declareProperty(jobKeyProp)
         .declareProperty(multiInstanceLoopCounterProp)
         .declareProperty(interruptingEventKeyProp)
@@ -123,6 +126,14 @@ public int getNumberOfActiveElementInstances() {
     return childCountProp.getValue();
   }
 
+  public int getNumberOfElementInstances() {
+    return childActivatedCountProp.getValue();
+  }
+
+  public void incrementNumberOfElementInstances() {
+    childActivatedCountProp.increment();
+  }
+
   public int getMultiInstanceLoopCounter() {
     return multiInstanceLoopCounterProp.getValue();
   }

engine/src/test/java/io/camunda/zeebe/engine/processing/bpmn/multiinstance/MultiInstanceActivityTest.java

@@ -1013,6 +1013,132 @@ public void shouldSetLoopCounterVariable() {
             tuple(elementInstanceKeys.get(2), "3"));
   }
 
+  @Test
+  public void shouldCompleteBodyWhenCompleteConditionAccessNumberOfInstancesEvaluateTrue() {
+    // given
+    int completedJobs = 2;
+    String completionCondition = "= numberOfInstances = 2";
+    if ("parallel".equals(loopCharacteristics)) {
+      completionCondition = "= numberOfInstances = 3";
+      completedJobs = 1;
+    }
+    final String finalCompletionCondition = completionCondition;
+    ENGINE
+        .deployment()
+        .withXmlResource(
+            process(miBuilder.andThen(m -> m.completionCondition(finalCompletionCondition))))
+        .deploy();
+
+    // when
+    final long processInstanceKey =
+        ENGINE
+            .processInstance()
+            .ofBpmnProcessId(PROCESS_ID)
+            .withVariable(INPUT_COLLECTION_EXPRESSION, INPUT_COLLECTION)
+            .create();
+
+    completeJobs(processInstanceKey, completedJobs);
+
+    // then
+    assertThat(
+            RecordingExporter.processInstanceRecords()
+                .withProcessInstanceKey(processInstanceKey)
+                .limitToProcessInstanceCompleted()
+                .withElementId(ELEMENT_ID))
+        .extracting(r -> tuple(r.getValue().getBpmnElementType(), r.getIntent()))
+        .containsSubsequence(
+            tuple(BpmnElementType.SERVICE_TASK, ProcessInstanceIntent.ELEMENT_COMPLETED),
+            tuple(BpmnElementType.MULTI_INSTANCE_BODY, ProcessInstanceIntent.COMPLETE_ELEMENT),
+            tuple(BpmnElementType.MULTI_INSTANCE_BODY, ProcessInstanceIntent.ELEMENT_COMPLETING),
+            tuple(BpmnElementType.MULTI_INSTANCE_BODY, ProcessInstanceIntent.ELEMENT_COMPLETED));
+
+    if ("parallel".equals(loopCharacteristics)) {
+      // after 1 has completed, the others must be terminated
+      assertThat(
+              RecordingExporter.records()
+                  .limitToProcessInstance(processInstanceKey)
+                  .processInstanceRecords()
+                  .withIntent(ProcessInstanceIntent.ELEMENT_TERMINATED)
+                  .withElementType(BpmnElementType.SERVICE_TASK)
+                  .count())
+          .describedAs("all non-completed service tasks have terminated")
+          .isEqualTo(2);
+    } else {
+      assertThat(
+              RecordingExporter.records()
+                  .limitToProcessInstance(processInstanceKey)
+                  .processInstanceRecords()
+                  .withIntent(ProcessInstanceIntent.ELEMENT_ACTIVATED)
+                  .withElementType(BpmnElementType.SERVICE_TASK)
+                  .count())
+          .describedAs("only 2 out of 3 sequential service tasks has activated")
+          .isEqualTo(2);
+    }
+  }
+
+  @Test
+  public void shouldCompleteBodyWhenCompleteConditionAccessNumberOfActiveInstancesEvaluateTrue() {
+    // given
+    int completedJobs = 1;
+    String completionCondition = "= numberOfActiveInstances = 0";
+    if ("parallel".equals(loopCharacteristics)) {
+      completedJobs = 2;
+      completionCondition = "= numberOfActiveInstances = 1";
+    }
+    final String finalCompletionCondition = completionCondition;
+    ENGINE
+        .deployment()
+        .withXmlResource(
+            process(miBuilder.andThen(m -> m.completionCondition(finalCompletionCondition))))
+        .deploy();
+
+    // when
+    final long processInstanceKey =
+        ENGINE
+            .processInstance()
+            .ofBpmnProcessId(PROCESS_ID)
+            .withVariable(INPUT_COLLECTION_EXPRESSION, INPUT_COLLECTION)
+            .create();
+
+    completeJobs(processInstanceKey, completedJobs);
+
+    // then
+    assertThat(
+            RecordingExporter.processInstanceRecords()
+                .withProcessInstanceKey(processInstanceKey)
+                .limitToProcessInstanceCompleted()
+                .withElementId(ELEMENT_ID))
+        .extracting(r -> tuple(r.getValue().getBpmnElementType(), r.getIntent()))
+        .containsSubsequence(
+            tuple(BpmnElementType.SERVICE_TASK, ProcessInstanceIntent.ELEMENT_COMPLETED),
+            tuple(BpmnElementType.MULTI_INSTANCE_BODY, ProcessInstanceIntent.COMPLETE_ELEMENT),
+            tuple(BpmnElementType.MULTI_INSTANCE_BODY, ProcessInstanceIntent.ELEMENT_COMPLETING),
+            tuple(BpmnElementType.MULTI_INSTANCE_BODY, ProcessInstanceIntent.ELEMENT_COMPLETED));
+
+    if ("parallel".equals(loopCharacteristics)) {
+      // after 2 has completed, the others must be terminated
+      assertThat(
+              RecordingExporter.records()
+                  .limitToProcessInstance(processInstanceKey)
+                  .processInstanceRecords()
+                  .withIntent(ProcessInstanceIntent.ELEMENT_TERMINATED)
+                  .withElementType(BpmnElementType.SERVICE_TASK)
+                  .count())
+          .describedAs("all non-completed service tasks have terminated")
+          .isEqualTo(1);
+    } else {
+      assertThat(
+              RecordingExporter.records()
+                  .limitToProcessInstance(processInstanceKey)
+                  .processInstanceRecords()
+                  .withIntent(ProcessInstanceIntent.ELEMENT_ACTIVATED)
+                  .withElementType(BpmnElementType.SERVICE_TASK)
+                  .count())
+          .describedAs("only 1 out of 3 sequential service tasks has activated")
+          .isEqualTo(1);
+    }
+  }
+
   @Test
   public void shouldApplyInputMapping() {
     // given