test(engine): verify StackOverFlow during termination results in banning

This test verifies that when a process instance is terminated and
results in a SO we will ban an instance.
It is not the process instance that got terminated that gets banned!
Instead it's one of the child instance for which the "bubbling up"
caused the SO.

(cherry picked from commit 83f22b0)

engine/src/test/java/io/camunda/zeebe/engine/processing/processinstance/CancelProcessInstanceBanTest.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH under
+ * one or more contributor license agreements. See the NOTICE file distributed
+ * with this work for additional information regarding copyright ownership.
+ * Licensed under the Zeebe Community License 1.1. You may not use this file
+ * except in compliance with the Zeebe Community License 1.1.
+ */
+package io.camunda.zeebe.engine.processing.processinstance;
+
+import io.camunda.zeebe.engine.util.EngineRule;
+import io.camunda.zeebe.model.bpmn.Bpmn;
+import io.camunda.zeebe.protocol.record.intent.ProcessInstanceIntent;
+import io.camunda.zeebe.protocol.record.value.BpmnElementType;
+import io.camunda.zeebe.test.util.Strings;
+import io.camunda.zeebe.test.util.record.RecordingExporter;
+import io.camunda.zeebe.test.util.record.RecordingExporterTestWatcher;
+import org.assertj.core.api.Assertions;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestWatcher;
+
+public final class CancelProcessInstanceBanTest {
+
+  @ClassRule public static final EngineRule ENGINE = EngineRule.singlePartition();
+  @Rule public final TestWatcher recordingExporterTestWatcher = new RecordingExporterTestWatcher();
+
+  @Test // Regression of https://github.com/camunda/zeebe/issues/8955
+  public void shouldBanInstanceWhenTerminatingInstanceWithALotOfNestedChildInstances() {
+    // given
+    final var amountOfNestedChildInstances = 1000;
+    final var processId = Strings.newRandomValidBpmnId();
+    ENGINE
+        .deployment()
+        .withXmlResource(
+            Bpmn.createExecutableProcess(processId)
+                .startEvent()
+                .exclusiveGateway()
+                .defaultFlow()
+                .userTask()
+                .endEvent()
+                .moveToLastGateway()
+                .conditionExpression("count < " + amountOfNestedChildInstances)
+                .intermediateThrowEvent("preventStraightThroughLoop")
+                .callActivity(
+                    "callActivity",
+                    c -> c.zeebeProcessId(processId).zeebeInputExpression("count + 1", "count"))
+                .endEvent()
+                .done())
+        .deploy();
+
+    final long processInstanceKey =
+        ENGINE.processInstance().ofBpmnProcessId(processId).withVariable("count", 0).create();
+
+    RecordingExporter.processInstanceRecords(ProcessInstanceIntent.ELEMENT_ACTIVATED)
+        .withElementType(BpmnElementType.USER_TASK)
+        .getFirst();
+
+    // when
+    final var errorRecordValue =
+        ENGINE.processInstance().withInstanceKey(processInstanceKey).cancelWithError();
+
+    // then
+    Assertions.assertThat(errorRecordValue.getValue().getStacktrace())
+        .contains("ChildTerminationStackOverflowException");
+    Assertions.assertThat(errorRecordValue.getValue().getExceptionMessage())
+        .contains(
+            "Process instance",
+            """
+            has too many nested child instances and could not be terminated. The deepest nested \
+            child instance has been banned as a result.""");
+  }
+}

engine/src/test/java/io/camunda/zeebe/engine/util/client/ProcessInstanceClient.java

@@ -20,9 +20,11 @@
 import io.camunda.zeebe.protocol.impl.record.value.processinstance.ProcessInstanceModificationVariableInstruction;
 import io.camunda.zeebe.protocol.impl.record.value.processinstance.ProcessInstanceRecord;
 import io.camunda.zeebe.protocol.record.Record;
+import io.camunda.zeebe.protocol.record.intent.ErrorIntent;
 import io.camunda.zeebe.protocol.record.intent.ProcessInstanceCreationIntent;
 import io.camunda.zeebe.protocol.record.intent.ProcessInstanceIntent;
 import io.camunda.zeebe.protocol.record.intent.ProcessInstanceModificationIntent;
+import io.camunda.zeebe.protocol.record.value.ErrorRecordValue;
 import io.camunda.zeebe.protocol.record.value.ProcessInstanceCreationRecordValue;
 import io.camunda.zeebe.protocol.record.value.ProcessInstanceModificationRecordValue;
 import io.camunda.zeebe.protocol.record.value.ProcessInstanceRecordValue;
@@ -197,6 +199,10 @@ public static class ExistingInstanceClient {
                 .withProcessInstanceKey(processInstanceKey)
                 .getFirst();
 
+    public static final Function<Long, Record<ErrorRecordValue>> ERROR_EXPECTATION =
+        (processInstanceKey) ->
+            RecordingExporter.errorRecords().withIntent(ErrorIntent.CREATED).getFirst();
+
     private static final int DEFAULT_PARTITION = -1;
     private final StreamProcessorRule environmentRule;
     private final long processInstanceKey;
@@ -221,6 +227,16 @@ public ExistingInstanceClient expectRejection() {
     }
 
     public Record<ProcessInstanceRecordValue> cancel() {
+      writeCancelCommand();
+      return expectation.apply(processInstanceKey);
+    }
+
+    public Record<ErrorRecordValue> cancelWithError() {
+      writeCancelCommand();
+      return ERROR_EXPECTATION.apply(processInstanceKey);
+    }
+
+    private void writeCancelCommand() {
       if (partition == DEFAULT_PARTITION) {
         partition =
             RecordingExporter.processInstanceRecords()
@@ -234,8 +250,6 @@ public Record<ProcessInstanceRecordValue> cancel() {
           processInstanceKey,
           ProcessInstanceIntent.CANCEL,
           new ProcessInstanceRecord().setProcessInstanceKey(processInstanceKey));
-
-      return expectation.apply(processInstanceKey);
     }
 
     public ProcessInstanceModificationClient modification() {

test-util/src/main/java/io/camunda/zeebe/test/util/record/ErrorRecordStream.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH under
+ * one or more contributor license agreements. See the NOTICE file distributed
+ * with this work for additional information regarding copyright ownership.
+ * Licensed under the Zeebe Community License 1.1. You may not use this file
+ * except in compliance with the Zeebe Community License 1.1.
+ */
+package io.camunda.zeebe.test.util.record;
+
+import io.camunda.zeebe.protocol.record.Record;
+import io.camunda.zeebe.protocol.record.value.ErrorRecordValue;
+import java.util.stream.Stream;
+
+public class ErrorRecordStream extends ExporterRecordStream<ErrorRecordValue, ErrorRecordStream> {
+
+  public ErrorRecordStream(final Stream<Record<ErrorRecordValue>> wrappedStream) {
+    super(wrappedStream);
+  }
+
+  @Override
+  protected ErrorRecordStream supply(final Stream<Record<ErrorRecordValue>> wrappedStream) {
+    return new ErrorRecordStream(wrappedStream);
+  }
+}

test-util/src/main/java/io/camunda/zeebe/test/util/record/RecordingExporter.java

@@ -33,6 +33,7 @@
 import io.camunda.zeebe.protocol.record.value.DecisionEvaluationRecordValue;
 import io.camunda.zeebe.protocol.record.value.DeploymentDistributionRecordValue;
 import io.camunda.zeebe.protocol.record.value.DeploymentRecordValue;
+import io.camunda.zeebe.protocol.record.value.ErrorRecordValue;
 import io.camunda.zeebe.protocol.record.value.EscalationRecordValue;
 import io.camunda.zeebe.protocol.record.value.IncidentRecordValue;
 import io.camunda.zeebe.protocol.record.value.JobBatchRecordValue;
@@ -344,6 +345,10 @@ public static ResourceDeletionRecordStream resourceDeletionRecords(
     return resourceDeletionRecords().withIntent(intent);
   }
 
+  public static ErrorRecordStream errorRecords() {
+    return new ErrorRecordStream(records(ValueType.ERROR, ErrorRecordValue.class));
+  }
+
   public static class AwaitingRecordIterator implements Iterator<Record<?>> {
 
     private int nextIndex = 0;