test(engine): make sure visit returns correct result

There is a bug in the visitMessagesWithDeadlineBeforeTimestamp, because
it wrongly informs the caller that the visitor commanded to stop the
visiting, while that is not the case. This can happen when there are
messages available that are not yet expired.

This test case will fail at this time.

engine/src/test/java/io/camunda/zeebe/engine/state/message/MessageStateTest.java

@@ -183,6 +183,59 @@ public void shouldNotVisitMessageIfCorrelationKeyDoesntMatch() {
     assertThat(keys).isEmpty();
   }
 
+  @Test
+  public void shouldVisitMessagesUntilVisitorReturnsFalse() {
+    // given
+    final var message = createMessage("name", "correlationKey", "{}", "nr1", 1234);
+    final var message2 = createMessage("otherName", "correlationKey", "{}", "nr2", 2000);
+
+    messageState.put(1L, message);
+    messageState.put(2L, message2);
+
+    // then
+    final List<Long> readMessage = new ArrayList<>();
+    final boolean isStoppedByVisitor =
+        messageState.visitMessagesWithDeadlineBeforeTimestamp(
+            3456,
+            null,
+            (deadline, e) -> {
+              readMessage.add(e);
+              return false;
+            });
+
+    assertThat(readMessage).hasSize(1).containsExactly(1L);
+    assertThat(isStoppedByVisitor)
+        .describedAs("Expect that the visiting stopped because of the visitor")
+        .isTrue();
+  }
+
+  @Test
+  public void shouldVisitMessagesWhileVisitorReturnsTrue() {
+    // given
+    final var message = createMessage("name", "correlationKey", "{}", "nr1", 1234);
+    final var message2 = createMessage("otherName", "correlationKey", "{}", "nr2", 2000);
+
+    messageState.put(1L, message);
+    messageState.put(2L, message2);
+
+    // then
+    final List<Long> readMessage = new ArrayList<>();
+    final boolean isStoppedByVisitor =
+        messageState.visitMessagesWithDeadlineBeforeTimestamp(
+            1_900,
+            null,
+            (deadline, e) -> {
+              readMessage.add(e);
+              return true;
+            });
+
+    assertThat(readMessage).hasSize(1).containsExactly(1L);
+    assertThat(isStoppedByVisitor)
+        .describedAs(
+            "Expect that the visiting is stopped because there are no more entries before the timestamp")
+        .isFalse();
+  }
+
   @Test
   public void shouldNotVisitMessagesBeforeTime() {
     // given