merge: #597

597: feat(filters): add event type for process instance record r=remcowesterhoud a=skayliu

## Description

Like `BpmnElementType`, `BpmnEventType` has been introduced to Zeebe SNAPSHOT for process instance record. so that as a consumer of the record stream, you can distinguish the type of event that is processed.


## Related issues

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

closes #596 

<!-- Cut-off marker
_All lines under and including the cut-off marker will be removed from the merge commit message_

## Definition of Ready

Please check the items that apply, before requesting a review.

You can find more details about these items in our wiki page about [Pull Requests and Code Reviews](https://github.com/camunda-cloud/zeebe/wiki/Pull-Requests-and-Code-Reviews).

* [ ] I've reviewed my own code
* [ ] I've written a clear changelist description
* [ ] I've narrowly scoped my changes
* [ ] I've separated structural from behavioural changes
-->

## Definition of Done

<!-- Please check the items that apply, before merging or (if possible) before requesting a review. -->

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

Code changes:
* [ ] The changes are backwards compatibility with previous versions
* [ ] If it fixes a bug then PRs are created to backport the fix

Testing:
* [x] There are unit/integration tests that verify all acceptance criterias of the issue
* [ ] New tests are written to ensure backwards compatibility with further versions
* [x] The behavior is tested manually

Documentation:
* [ ] Javadoc has been written
* [ ] The documentation is updated


Co-authored-by: skayliu <skay463@163.com>

engine/src/test/java/io/camunda/zeebe/process/test/engine/EngineClientTest.java

@@ -34,6 +34,7 @@
 import io.camunda.zeebe.protocol.record.intent.ProcessInstanceIntent;
 import io.camunda.zeebe.protocol.record.intent.TimerIntent;
 import io.camunda.zeebe.protocol.record.value.BpmnElementType;
+import io.camunda.zeebe.protocol.record.value.BpmnEventType;
 import io.camunda.zeebe.protocol.record.value.JobRecordValue;
 import io.camunda.zeebe.protocol.record.value.ProcessInstanceRecordValue;
 import io.camunda.zeebe.protocol.record.value.TimerRecordValue;
@@ -328,6 +329,7 @@ void shouldCancelProcessInstance() {
                           false)
                       .filter(r -> r.getIntent() == ProcessInstanceIntent.ELEMENT_TERMINATED)
                       .filter(r -> r.getValue().getBpmnElementType() == BpmnElementType.PROCESS)
+                      .filter(r -> r.getValue().getBpmnEventType() == BpmnEventType.UNSPECIFIED)
                       .findFirst();
               assertThat(first).isNotEmpty();
             });
@@ -699,6 +701,7 @@ void shouldThrowErrorOnJob() {
                                     r ->
                                         r.getValue().getBpmnElementType()
                                             == BpmnElementType.BOUNDARY_EVENT)
+                                .filter(r -> r.getValue().getBpmnEventType() == BpmnEventType.ERROR)
                                 .findFirst();
 
                         assertThat(boundaryEvent).isNotEmpty();
@@ -800,6 +803,7 @@ void shouldReadProcessInstanceRecords() {
                           false)
                       .filter(r -> r.getRecordType() == RecordType.EVENT)
                       .filter(r -> r.getValue().getBpmnElementType() == BpmnElementType.PROCESS)
+                      .filter(r -> r.getValue().getBpmnEventType() == BpmnEventType.UNSPECIFIED)
                       .limit(4)
                       .collect(Collectors.toList());
 
@@ -822,6 +826,8 @@ void shouldReadProcessInstanceRecords() {
                   .isEqualTo(processInstance.getVersion());
               assertThat(processInstanceRecord.getBpmnElementType())
                   .isEqualTo(BpmnElementType.PROCESS);
+              assertThat(processInstanceRecord.getBpmnEventType())
+                  .isEqualTo(BpmnEventType.UNSPECIFIED);
             });
   }
 
@@ -853,6 +859,7 @@ void shouldPrintRecords() {
               final Optional<Record<ProcessInstanceRecordValue>> processRecords =
                   StreamSupport.stream(recordStream.processInstanceRecords().spliterator(), false)
                       .filter(r -> r.getValue().getBpmnElementType() == BpmnElementType.PROCESS)
+                      .filter(r -> r.getValue().getBpmnEventType() == BpmnEventType.UNSPECIFIED)
                       .filter(r -> r.getIntent() == ProcessInstanceIntent.ELEMENT_COMPLETED)
                       .findFirst();
 
@@ -908,6 +915,7 @@ void shouldIncreaseTheTime() {
                               .spliterator(),
                           false)
                       .filter(r -> r.getValue().getBpmnElementType() == BpmnElementType.PROCESS)
+                      .filter(r -> r.getValue().getBpmnEventType() == BpmnEventType.UNSPECIFIED)
                       .filter(r -> r.getIntent() == ProcessInstanceIntent.ELEMENT_COMPLETED)
                       .findFirst();
 

filters/src/main/java/io/camunda/zeebe/process/test/filters/ProcessInstanceRecordStreamFilter.java

@@ -20,6 +20,7 @@
 import io.camunda.zeebe.protocol.record.RejectionType;
 import io.camunda.zeebe.protocol.record.intent.ProcessInstanceIntent;
 import io.camunda.zeebe.protocol.record.value.BpmnElementType;
+import io.camunda.zeebe.protocol.record.value.BpmnEventType;
 import io.camunda.zeebe.protocol.record.value.ProcessInstanceRecordValue;
 import java.util.Arrays;
 import java.util.stream.Stream;
@@ -56,6 +57,16 @@ public ProcessInstanceRecordStreamFilter withoutBpmnElementType(
         stream.filter(record -> record.getValue().getBpmnElementType() != bpmnElementType));
   }
 
+  public ProcessInstanceRecordStreamFilter withBpmnEventType(final BpmnEventType bpmnEventType) {
+    return new ProcessInstanceRecordStreamFilter(
+        stream.filter(record -> record.getValue().getBpmnEventType() == bpmnEventType));
+  }
+
+  public ProcessInstanceRecordStreamFilter withoutBpmnEventType(final BpmnEventType bpmnEventType) {
+    return new ProcessInstanceRecordStreamFilter(
+        stream.filter(record -> record.getValue().getBpmnEventType() != bpmnEventType));
+  }
+
   public ProcessInstanceRecordStreamFilter withIntent(final ProcessInstanceIntent intent) {
     return new ProcessInstanceRecordStreamFilter(
         stream.filter(record -> record.getIntent() == intent));

filters/src/main/java/io/camunda/zeebe/process/test/filters/logger/RecordStreamLogger.java

@@ -163,6 +163,7 @@ private String logProcessInstanceRecordValue(final Record<?> record) {
     final StringJoiner joiner = new StringJoiner(", ", "", "");
     joiner.add(String.format("(Element id: %s)", value.getElementId()));
     joiner.add(String.format("(Element type: %s)", value.getBpmnElementType()));
+    joiner.add(String.format("(Event type: %s)", value.getBpmnEventType()));
     joiner.add(String.format("(Process id: %s)", value.getBpmnProcessId()));
     return joiner.toString();
   }