merge: #10789

10789: Reset active sequence flows upon terminate end event activation r=remcowesterhoud a=remcowesterhoud

## Description

<!-- Please explain the changes you made here. -->
After a terminate end event has been completed the flow scope should be completed as well. When there are active sequence flows this doesn't happen as we have checks preventing a flow scope from completing when there is any active element instance or active sequence flow. By resetting this on activation of the terminate end event we can pass this check.

## Related issues

<!-- Which issues are closed by this PR or are related -->

closes #10590 



Co-authored-by: Remco Westerhoud <remco@westerhoud.nl>

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

@@ -91,8 +91,6 @@ private void cleanupSequenceFlowsTaken(final ProcessInstanceRecord value) {
       // (Tetris principle)
       elementInstanceState.decrementNumberOfTakenSequenceFlows(
           value.getFlowScopeKey(), gateway.getId());
-    } else {
-      return;
     }
   }
 

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

@@ -8,6 +8,7 @@
 package io.camunda.zeebe.engine.state.appliers;
 
 import io.camunda.zeebe.engine.processing.deployment.model.element.ExecutableCallActivity;
+import io.camunda.zeebe.engine.processing.deployment.model.element.ExecutableEndEvent;
 import io.camunda.zeebe.engine.state.TypedEventApplier;
 import io.camunda.zeebe.engine.state.immutable.ProcessState;
 import io.camunda.zeebe.engine.state.instance.ElementInstance;
@@ -63,6 +64,11 @@ public void applyState(final long key, final ProcessInstanceRecord value) {
 
     final var flowScopeElementType = flowScopeInstance.getValue().getBpmnElementType();
     manageMultiInstance(flowScopeInstance, flowScopeElementType);
+
+    if (isTerminateEndEvent(value)) {
+      flowScopeInstance.resetActiveSequenceFlows();
+      elementInstanceState.updateInstance(flowScopeInstance);
+    }
   }
 
   private boolean isChildProcess(
@@ -97,4 +103,16 @@ private void manageMultiInstance(
       elementInstanceState.updateInstance(flowScopeInstance);
     }
   }
+
+  private boolean isTerminateEndEvent(final ProcessInstanceRecord value) {
+    if (value.getBpmnElementType().equals(BpmnElementType.END_EVENT)) {
+      final var element =
+          processState.getFlowElement(
+              value.getProcessDefinitionKey(),
+              value.getElementIdBuffer(),
+              ExecutableEndEvent.class);
+      return element.isTerminateEndEvent();
+    }
+    return false;
+  }
 }

engine/src/test/java/io/camunda/zeebe/engine/processing/bpmn/activity/TerminateEndEventTest.java

@@ -476,4 +476,47 @@ record -> record.getValue().getElementId(),
                 BpmnElementType.END_EVENT, "end_after_C", ProcessInstanceIntent.ELEMENT_COMPLETED),
             tuple(BpmnElementType.PROCESS, PROCESS_ID, ProcessInstanceIntent.ELEMENT_COMPLETED));
   }
+
+  @Test
+  public void shouldCompleteProcessWhenWaitingAtParallelGateway() {
+    ENGINE_RULE
+        .deployment()
+        .withXmlResource(
+            Bpmn.createExecutableProcess(PROCESS_ID)
+                .startEvent()
+                .parallelGateway("fork")
+                .parallelGateway("join")
+                .moveToNode("fork")
+                .serviceTask("A", s -> s.zeebeJobType("type"))
+                .boundaryEvent()
+                .error("code")
+                .endEvent()
+                .terminate()
+                .moveToNode("A")
+                .connectTo("join")
+                .endEvent()
+                .done())
+        .deploy();
+
+    final long processInstanceKey =
+        ENGINE_RULE.processInstance().ofBpmnProcessId(PROCESS_ID).create();
+
+    ENGINE_RULE
+        .job()
+        .ofInstance(processInstanceKey)
+        .withType("type")
+        .withErrorCode("code")
+        .throwError();
+
+    Assertions.assertThat(
+            RecordingExporter.processInstanceRecords()
+                .withProcessInstanceKey(processInstanceKey)
+                .limitToProcessInstanceCompleted())
+        .extracting(record -> record.getValue().getBpmnElementType(), Record::getIntent)
+        .describedAs(
+            "Expect to terminate all element instances when reaching the terminate end event")
+        .containsSubsequence(
+            tuple(BpmnElementType.END_EVENT, ProcessInstanceIntent.ELEMENT_COMPLETED),
+            tuple(BpmnElementType.PROCESS, ProcessInstanceIntent.ELEMENT_COMPLETED));
+  }
 }