test(engine): cancel jobs with incidents

This adds a regression test against activating jobs of cancelled process
instances, by making sure they are cancelled (i.e. deleted).

In this specific case, an uncaught error was thrown on a job which led
to an UNHANDLED_ERROR_EVENT incident. When the process instance is
cancelled, the job was not called, but instead was made activatable
again. This happened, because the incident was resolved as part of the
process instance termination logic. Instead, we should guarantee that
the job is cancelled. A cancelled job can no longer be made
re-activatable. This is a terminal state for jobs, because a cancelled
job is a deleted job.

See #8588 for more details.

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

@@ -21,6 +21,7 @@
 import io.camunda.zeebe.protocol.record.Record;
 import io.camunda.zeebe.protocol.record.RecordType;
 import io.camunda.zeebe.protocol.record.RejectionType;
+import io.camunda.zeebe.protocol.record.intent.IncidentIntent;
 import io.camunda.zeebe.protocol.record.intent.JobIntent;
 import io.camunda.zeebe.protocol.record.intent.ProcessInstanceIntent;
 import io.camunda.zeebe.protocol.record.value.BpmnElementType;
@@ -447,4 +448,40 @@ public void shouldWriteEntireEventOnCancel() {
     // then
     assertThat(canceledRecord.getValue()).isEqualTo(activatedEvent.getValue());
   }
+
+  /**
+   * Regression test against activating jobs of cancelled process instances
+   *
+   * <p>See: https://github.com/camunda/zeebe/issues/8588
+   */
+  @Test
+  public void shouldCancelJobsWithIncidents() {
+    // given
+    final var processInstanceKey = ENGINE.processInstance().ofBpmnProcessId("PROCESS").create();
+    final var job = ENGINE.job().ofInstance(processInstanceKey).withType("test").throwError();
+    assertThat(
+            RecordingExporter.incidentRecords(IncidentIntent.CREATED)
+                .withProcessInstanceKey(processInstanceKey)
+                .withJobKey(job.getKey())
+                .findAny())
+        .describedAs("Expect an incident on the job")
+        .isPresent();
+
+    // when
+    ENGINE.processInstance().withInstanceKey(processInstanceKey).cancel();
+    assertThat(
+            RecordingExporter.processInstanceRecords()
+                .withElementType(BpmnElementType.PROCESS)
+                .withProcessInstanceKey(processInstanceKey)
+                .limitToProcessInstanceTerminated()
+                .findAny())
+        .describedAs("Wait until the process instance has terminated")
+        .isPresent();
+
+    // then
+    assertThat(
+            RecordingExporter.jobRecords(JobIntent.CANCELED).withRecordKey(job.getKey()).findAny())
+        .describedAs("Expect that the job is cancelled")
+        .isPresent();
+  }
 }