-
Notifications
You must be signed in to change notification settings - Fork 556
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
9219: Cancel job with incident when canceling the process instance r=korthout a=korthout ## Description This fixes a bug where canceling a process instance could lead to a job becoming activatable even though the process instance is terminated. See #8588 for more details. The bug is fixed by making sure the job is immediately canceled (i.e. deleted from the state) when we terminate the related element instance. We achieve this by writing a Job Canceled event when the TERMINATE_ELEMENT command is being processed. This PR fixes the bug in a way that can be backported because it doesn't introduce too many changes. Alternatively, the bug could be fixed by tackling the main issue: applying the Incident Resolved event should not affect the state of a job. Instead, the job should be made activatable by writing a separate Job event in the Resolve Incident Processor. That would be a larger change, with more impact on the event appliers than what this PR proposes. Changes to event appliers should generally be avoided. ## Related issues <!-- Which issues are closed by this PR or are related --> closes #8588 Co-authored-by: Nico Korthout <nico.korthout@camunda.com>
- Loading branch information
Showing
9 changed files
with
285 additions
and
35 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
187 changes: 187 additions & 0 deletions
187
engine/src/test/java/io/camunda/zeebe/engine/metrics/JobMetricsTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,187 @@ | ||
| /* | ||
| * 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.metrics; | ||
|
|
||
| import static java.util.Map.entry; | ||
| import static org.assertj.core.api.Assertions.assertThat; | ||
|
|
||
| import io.camunda.zeebe.engine.util.EngineRule; | ||
| import io.camunda.zeebe.model.bpmn.Bpmn; | ||
| import io.camunda.zeebe.protocol.record.intent.JobIntent; | ||
| import io.camunda.zeebe.test.util.record.RecordingExporter; | ||
| import io.camunda.zeebe.test.util.record.RecordingExporterTestWatcher; | ||
| import java.time.Duration; | ||
| import org.junit.Before; | ||
| import org.junit.BeforeClass; | ||
| import org.junit.ClassRule; | ||
| import org.junit.Rule; | ||
| import org.junit.Test; | ||
| import org.junit.rules.TestWatcher; | ||
|
|
||
| public class JobMetricsTest { | ||
|
|
||
| @ClassRule public static final EngineRule ENGINE = EngineRule.singlePartition(); | ||
|
|
||
| private static final String PROCESS_ID = "process"; | ||
| private static final String TASK_ID = "task"; | ||
| private static final String JOB_TYPE = "job"; | ||
|
|
||
| @Rule public final TestWatcher watcher = new RecordingExporterTestWatcher(); | ||
|
|
||
| @BeforeClass | ||
| public static void deployProcess() { | ||
| ENGINE | ||
| .deployment() | ||
| .withXmlResource( | ||
| Bpmn.createExecutableProcess(PROCESS_ID) | ||
| .startEvent() | ||
| .serviceTask(TASK_ID, t -> t.zeebeJobTypeExpression("jobType")) | ||
| .endEvent() | ||
| .done()) | ||
| .deploy(); | ||
| } | ||
|
|
||
| @Before | ||
| public void resetMetrics() { | ||
| JobMetrics.clear(); | ||
| } | ||
|
|
||
| @Test | ||
| public void allCountsStartAtNull() { | ||
| assertThat(jobMetric("created", JOB_TYPE)).isNull(); | ||
| assertThat(jobMetric("activated", JOB_TYPE)).isNull(); | ||
| assertThat(jobMetric("timed out", JOB_TYPE)).isNull(); | ||
| assertThat(jobMetric("completed", JOB_TYPE)).isNull(); | ||
| assertThat(jobMetric("failed", JOB_TYPE)).isNull(); | ||
| assertThat(jobMetric("canceled", JOB_TYPE)).isNull(); | ||
| assertThat(jobMetric("error thrown", JOB_TYPE)).isNull(); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldCountCreated() { | ||
| // when | ||
| createProcessInstanceWithJob(JOB_TYPE); | ||
|
|
||
| // then | ||
| assertThat(jobMetric("created", JOB_TYPE)).isNotNull().isEqualTo(1); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldCountActivated() { | ||
| // given | ||
|
|
||
| // the job type must be unique, because other tests may also have created jobs that can be | ||
| // activated. We can't depend on the unique process instance when activating a batch of jobs. | ||
| final String jobType = JOB_TYPE + "_activated"; | ||
| createProcessInstanceWithJob(jobType); | ||
|
|
||
| // when | ||
| ENGINE.jobs().withType(jobType).activate(); | ||
|
|
||
| // then | ||
| assertThat(jobMetric("activated", jobType)).isNotNull().isEqualTo(1); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldCountTimedOut() { | ||
| // given | ||
| final long processInstanceKey = createProcessInstanceWithJob(JOB_TYPE); | ||
|
|
||
| final var timeout = Duration.ofMinutes(10); | ||
| ENGINE.jobs().withType(JOB_TYPE).withTimeout(timeout.toMillis()).activate(); | ||
|
|
||
| // when | ||
| ENGINE.getClock().addTime(timeout); | ||
| RecordingExporter.jobRecords(JobIntent.TIMED_OUT) | ||
| .withProcessInstanceKey(processInstanceKey) | ||
| .await(); | ||
|
|
||
| // then | ||
| assertThat(jobMetric("timed out", JOB_TYPE)).isNotNull().isEqualTo(1); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldCountCompleted() { | ||
| // given | ||
| final long processInstanceKey = createProcessInstanceWithJob(JOB_TYPE); | ||
|
|
||
| // when | ||
| ENGINE.job().ofInstance(processInstanceKey).withType(JOB_TYPE).complete(); | ||
|
|
||
| // then | ||
| assertThat(jobMetric("completed", JOB_TYPE)).isNotNull().isEqualTo(1); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldCountFailed() { | ||
| // given | ||
| final long processInstanceKey = createProcessInstanceWithJob(JOB_TYPE); | ||
|
|
||
| // when | ||
| ENGINE.job().ofInstance(processInstanceKey).withType(JOB_TYPE).fail(); | ||
|
|
||
| // then | ||
| assertThat(jobMetric("failed", JOB_TYPE)).isNotNull().isEqualTo(1); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldCountCanceled() { | ||
| // given | ||
| final long processInstanceKey = createProcessInstanceWithJob(JOB_TYPE); | ||
|
|
||
| // when | ||
| ENGINE.processInstance().withInstanceKey(processInstanceKey).cancel(); | ||
| RecordingExporter.jobRecords(JobIntent.CANCELED) | ||
| .withProcessInstanceKey(processInstanceKey) | ||
| .await(); | ||
|
|
||
| // then | ||
| assertThat(jobMetric("canceled", JOB_TYPE)).isNotNull().isEqualTo(1); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldCountErrorThrown() { | ||
| // given | ||
| final long processInstanceKey = createProcessInstanceWithJob(JOB_TYPE); | ||
|
|
||
| // when | ||
| ENGINE.job().ofInstance(processInstanceKey).withType(JOB_TYPE).throwError(); | ||
|
|
||
| // then | ||
| assertThat(jobMetric("error thrown", JOB_TYPE)).isNotNull().isEqualTo(1); | ||
| } | ||
|
|
||
| /** | ||
| * Creates a process instance with a job, and waits until the job is created | ||
| * | ||
| * @param jobType the job type for the service task | ||
| * @return the key of the created process instance | ||
| */ | ||
| private static long createProcessInstanceWithJob(final String jobType) { | ||
| final long processInstanceKey = | ||
| ENGINE | ||
| .processInstance() | ||
| .ofBpmnProcessId(PROCESS_ID) | ||
| .withVariable("jobType", jobType) | ||
| .create(); | ||
|
|
||
| RecordingExporter.jobRecords(JobIntent.CREATED) | ||
| .withProcessInstanceKey(processInstanceKey) | ||
| .await(); | ||
|
|
||
| return processInstanceKey; | ||
| } | ||
|
|
||
| private static Double jobMetric(final String action, final String type) { | ||
| return MetricsTestHelper.readMetricValue( | ||
| "zeebe_job_events_total", | ||
| entry("action", action), | ||
| entry("partition", "1"), | ||
| entry("type", type)); | ||
| } | ||
| } |
34 changes: 34 additions & 0 deletions
34
engine/src/test/java/io/camunda/zeebe/engine/metrics/MetricsTestHelper.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| /* | ||
| * 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.metrics; | ||
|
|
||
| import io.prometheus.client.CollectorRegistry; | ||
| import java.util.Arrays; | ||
| import java.util.List; | ||
| import java.util.Map.Entry; | ||
|
|
||
| final class MetricsTestHelper { | ||
|
|
||
| /** | ||
| * Reads the value of a metric based on its name and labels. | ||
| * | ||
| * <p>This uses the defaultRegistry which is inefficient, so this should only be used in testing. | ||
| * | ||
| * @param name the name of the metric | ||
| * @param labels names and values for labels. This is useful for filtering the right value within | ||
| * the metric. | ||
| * @return the given value or null if it doesn't exist | ||
| */ | ||
| @SafeVarargs | ||
| static Double readMetricValue(final String name, final Entry<String, String>... labels) { | ||
| final List<String> labelNames = Arrays.stream(labels).map(Entry::getKey).toList(); | ||
| final List<String> labelValues = Arrays.stream(labels).map(Entry::getValue).toList(); | ||
| return CollectorRegistry.defaultRegistry.getSampleValue( | ||
| name, labelNames.toArray(new String[] {}), labelValues.toArray(new String[] {})); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.