merge: #10054

10054: Avoid timer cycle with start time triggering multiple times r=korthout a=lzgabel

## Description
If a repeating timer with a start time is defined, the engine is stopped for some time during which the timer is supposed to trigger multiple times, and the engine is restarted, then it should trigger only once and reschedule the next timer in the future. Also, if a past start time is defined, it triggers after one interval period. If a future start time is defined that is expected sooner than, at, or after one interval period, it triggers at the start time.

<!-- Please explain the changes you made here. -->

## Related issues

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

closes #9680 



Co-authored-by: lzgabel <lz19960321lz@gmail.com>

bpmn-model/src/main/java/io/camunda/zeebe/model/bpmn/util/time/Interval.java

@@ -75,7 +75,10 @@ public Optional<ZonedDateTime> getStart() {
   }
 
   public long toEpochMilli(final long fromEpochMilli) {
-    if (!start.isPresent()) {
+    final long epochMilli =
+        start.map(ZonedDateTime::toInstant).map(Instant::toEpochMilli).orElse(fromEpochMilli);
+
+    if (epochMilli <= fromEpochMilli) {
       if (!isCalendarBased()) {
         return fromEpochMilli + getDuration().toMillis();
       }
@@ -86,7 +89,7 @@ public long toEpochMilli(final long fromEpochMilli) {
           .toEpochMilli();
     }
 
-    return start.get().toInstant().toEpochMilli();
+    return epochMilli;
   }
 
   /**

bpmn-model/src/test/java/io/camunda/zeebe/model/bpmn/util/time/RepeatingIntervalTest.java

@@ -198,4 +198,21 @@ public void shouldCalculateDueDate() {
     // then
     assertThat(newDueDate).isEqualTo(expected);
   }
+
+  @Test
+  public void shouldNotBeLessThanCurrentTime() {
+    // given
+    final Interval interval = new Interval(Period.ZERO, Duration.ofSeconds(10));
+    final Instant currentTime = Instant.now();
+    final long fromEpochMilli = currentTime.toEpochMilli();
+
+    // set start time to be less than current time
+    final Instant start = currentTime.minus(Duration.ofDays(1));
+
+    // when
+    final long dueDate = interval.withStart(start).toEpochMilli(fromEpochMilli);
+
+    // then
+    assertThat(dueDate).isEqualTo(currentTime.plusSeconds(10).toEpochMilli());
+  }
 }

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

@@ -1071,7 +1071,7 @@ public void shouldTriggerOnlyTwice() {
     final long secondDeploymentProcessDefinitionKey = secondDeployment.getProcessDefinitionKey();
 
     // when
-    engine.increaseTime(Duration.ofSeconds(25));
+    engine.increaseTime(Duration.ofSeconds(15));
 
     // then
     assertThat(
@@ -1242,4 +1242,56 @@ public void shouldCreateCronTimer() {
                 .exists())
         .isTrue();
   }
+
+  @Test
+  public void shouldAvoidTriggeringMultipleTimes() {
+    // given
+    final ZonedDateTime start =
+        ZonedDateTime.ofInstant(Instant.now(), ZoneId.systemDefault()).plusMinutes(30);
+    final BpmnModelInstance model =
+        Bpmn.createExecutableProcess("process")
+            .startEvent("start")
+            .timerWithCycle(String.format("R3/%s/PT10M", start))
+            .endEvent("end")
+            .done();
+    final var deployedProcess =
+        engine
+            .deployment()
+            .withXmlResource(model)
+            .deploy()
+            .getValue()
+            .getProcessesMetadata()
+            .get(0);
+    final long processDefinitionKey = deployedProcess.getProcessDefinitionKey();
+
+    // when
+    engine.stop();
+    final long engineStoppedTime = engine.getClock().getCurrentTimeInMillis();
+    engine.increaseTime(Duration.ofMinutes(35));
+    RecordingExporter.reset();
+    engine.start();
+
+    // then
+    final Record<TimerRecordValue> firstRecord =
+        RecordingExporter.timerRecords(TimerIntent.TRIGGERED)
+            .withProcessDefinitionKey(processDefinitionKey)
+            .getFirst();
+
+    Assertions.assertThat(firstRecord.getValue())
+        .hasDueDate(start.toInstant().toEpochMilli())
+        .hasTargetElementId("start")
+        .hasElementInstanceKey(TimerInstance.NO_ELEMENT_INSTANCE);
+
+    assertThat(firstRecord.getTimestamp()).isGreaterThan(engineStoppedTime);
+
+    final TimerRecordValue secondTimerRecord =
+        RecordingExporter.timerRecords(TimerIntent.CREATED)
+            .withProcessDefinitionKey(processDefinitionKey)
+            .skip(1)
+            .getFirst()
+            .getValue();
+
+    final long now = engine.getClock().getCurrentTimeInMillis();
+    assertThat(secondTimerRecord.getDueDate()).isGreaterThan(now);
+  }
 }