merge: #10893

10893: [Backport stable/8.0] Unflake timer start event test r=remcowesterhoud a=korthout

# Description
Backport of #10867 to `stable/8.0`.

relates to #10272

---

Reviewer, I had to resolve a few conflicts:
- some import changes
- new tests in TimerStartEventTest were changed, but these tests don't exist in 8.0

Co-authored-by: Nico Korthout <nico.korthout@camunda.com>
Co-authored-by: Remco Westerhoud <remco@westerhoud.nl>

engine/src/test/java/io/camunda/zeebe/engine/processing/timer/TimerCatchEventTest.java

@@ -204,8 +204,6 @@ public void shouldTriggerTimer() {
             .withProcessInstanceKey(processInstanceKey)
             .getFirst();
 
-    ENGINE.increaseTime(Duration.ofSeconds(1));
-
     // then
     final Record<TimerRecordValue> triggeredEvent =
         RecordingExporter.timerRecords(TimerIntent.TRIGGERED)
@@ -215,7 +213,7 @@ public void shouldTriggerTimer() {
     assertThat(triggeredEvent.getKey()).isEqualTo(createdEvent.getKey());
     assertThat(triggeredEvent.getValue()).isEqualTo(createdEvent.getValue());
     assertThat(Duration.ofMillis(triggeredEvent.getTimestamp() - createdEvent.getTimestamp()))
-        .isGreaterThanOrEqualTo(Duration.ofSeconds(1));
+        .isBetween(Duration.ofMillis(100), Duration.ofMillis(150));
   }
 
   @Test

engine/src/test/java/io/camunda/zeebe/engine/util/EngineRule.java

@@ -20,6 +20,7 @@
 import io.camunda.zeebe.engine.processing.streamprocessor.ReadonlyProcessingContext;
 import io.camunda.zeebe.engine.processing.streamprocessor.RecordValues;
 import io.camunda.zeebe.engine.processing.streamprocessor.StreamProcessor;
+import io.camunda.zeebe.engine.processing.streamprocessor.StreamProcessor.Phase;
 import io.camunda.zeebe.engine.processing.streamprocessor.StreamProcessorLifecycleAware;
 import io.camunda.zeebe.engine.processing.streamprocessor.StreamProcessorListener;
 import io.camunda.zeebe.engine.processing.streamprocessor.StreamProcessorMode;
@@ -242,6 +243,18 @@ public void forEachPartition(final Consumer<Integer> partitionIdConsumer) {
   }
 
   public void increaseTime(final Duration duration) {
+    final var streamProcessor = environmentRule.getStreamProcessor(PARTITION_ID);
+    if (streamProcessor.getCurrentPhase().join() == Phase.PROCESSING) {
+      // When time traveling, we're generally want to make sure that the entire state machine cycle
+      // for processing a record is completed, including the execution of post-commit tasks. For
+      // example, we're often interested in scheduled timers when time traveling in tests, for which
+      // the due date checker is scheduled through a post-commit task. When the engine has reached
+      // the end of the log, all post-commit tasks have also been applied, because the state machine
+      // will have executed them before switching the hasReachEnd flag.
+      Awaitility.await("Expect that engine reaches the end of the log before increasing the time")
+          .until(this::hasReachedEnd);
+    }
+
     environmentRule.getClock().addTime(duration);
   }