test(engine): verify message correlation to non-interrupting event

(cherry picked from commit a6cd7f4)

engine/src/test/java/io/camunda/zeebe/engine/processing/message/MessageCorrelationTest.java

@@ -831,6 +831,59 @@ public void shouldCorrelateToNonInterruptingBoundaryEvent() {
         .containsExactly("0", "1", "2");
   }
 
+  @Test
+  public void shouldCorrelateOnlyOnceToNonInterruptingBoundaryEvent() {
+    // given
+    final BpmnModelInstance process =
+        Bpmn.createExecutableProcess(PROCESS_ID)
+            .startEvent()
+            .serviceTask("task", b -> b.zeebeJobType("test"))
+            .boundaryEvent("message")
+            .cancelActivity(false)
+            .message(m -> m.name("message").zeebeCorrelationKeyExpression("key"))
+            .endEvent("correlated")
+            .moveToActivity("task")
+            .boundaryEvent("timer")
+            .timerWithDuration("PT5M")
+            .endEvent()
+            .done();
+
+    engine.deployment().withXmlResource(process).deploy();
+
+    final var processInstanceKey =
+        engine.processInstance().ofBpmnProcessId(PROCESS_ID).withVariable("key", "key-1").create();
+
+    // when
+    engine.message().withName("message").withCorrelationKey("key-1").publish();
+
+    RecordingExporter.processInstanceRecords(ProcessInstanceIntent.ELEMENT_COMPLETED)
+        .withProcessInstanceKey(processInstanceKey)
+        .withElementType(BpmnElementType.BOUNDARY_EVENT)
+        .await();
+
+    // complete the process instance by triggering the interrupting timer boundary event
+    engine.increaseTime(Duration.ofMinutes(5));
+
+    // then
+    assertThat(
+            RecordingExporter.processInstanceRecords()
+                .withProcessInstanceKey(processInstanceKey)
+                .limitToProcessInstanceCompleted()
+                .withIntent(ProcessInstanceIntent.ELEMENT_ACTIVATED)
+                .withElementType(BpmnElementType.BOUNDARY_EVENT))
+        .describedAs("Expected that the message boundary event is activated only once")
+        .hasSize(2)
+        .extracting(r -> r.getValue().getElementId())
+        .containsExactly("message", "timer");
+
+    assertThat(
+            RecordingExporter.records()
+                .limitToProcessInstance(processInstanceKey)
+                .onlyCommandRejections())
+        .describedAs("Expected no subscription command rejections")
+        .isEmpty();
+  }
+
   @Test
   public void shouldCorrelateMessageAgainAfterRejection() {
     // given