feat(broker-core): support non-interrupting timer boundary events

- timeDuration now also uses Interval (i.e. supports P1Y2DT3S)
- timers are recreated until they run out of repetitions or are canceled
- number of repetitions is encoded in TimerRecord/TimerInstance
- fixed token count with non-interrupting boundary events
- rename BoundaryEventHelper to BoundaryEventActivator
- introduce ExecutableCatchEventElement

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

@@ -16,18 +16,34 @@
 package io.zeebe.model.bpmn.util.time;
 
 import java.time.Duration;
+import java.time.Instant;
 import java.time.Period;
-import java.time.format.DateTimeParseException;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.time.temporal.ChronoUnit;
+import java.time.temporal.Temporal;
+import java.time.temporal.TemporalAmount;
+import java.time.temporal.TemporalUnit;
+import java.time.temporal.UnsupportedTemporalTypeException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Objects;
 
 /** Combines {@link java.time.Period}, and {@link java.time.Duration} */
-public class Interval {
+public class Interval implements TemporalAmount {
+  private static final Duration ACCURATE_DURATION_UPPER_BOUND = Duration.ofDays(1);
+
+  private final List<TemporalUnit> units;
   private final Period period;
   private final Duration duration;
 
   public Interval(Period period, Duration duration) {
     this.period = period;
     this.duration = duration;
+    this.units = new ArrayList<>();
+
+    this.units.addAll(period.getUnits());
+    this.units.addAll(duration.getUnits());
   }
 
   public Period getPeriod() {
@@ -38,6 +54,49 @@ public Duration getDuration() {
     return duration;
   }
 
+  public long toEpochMilli(long fromEpochMilli) {
+    if (!isCalendarBased()) {
+      return fromEpochMilli + getDuration().toMillis();
+    }
+
+    final Instant start = Instant.ofEpochMilli(fromEpochMilli);
+    final ZonedDateTime zoneAwareStart = ZonedDateTime.ofInstant(start, ZoneId.systemDefault());
+    return zoneAwareStart.plus(this).toInstant().toEpochMilli();
+  }
+
+  /**
+   * {@link Duration#get(TemporalUnit)} only accepts {@link ChronoUnit#SECONDS} and {@link
+   * ChronoUnit#NANOS}, so for any other units, this call is delegated to {@link
+   * Period#get(TemporalUnit)}, though it could easily be the other way around.
+   *
+   * @param unit the {@code TemporalUnit} for which to return the value
+   * @return the long value of the unit
+   * @throws UnsupportedTemporalTypeException if the unit is not supported
+   */
+  @Override
+  public long get(TemporalUnit unit) {
+    if (unit == ChronoUnit.SECONDS || unit == ChronoUnit.NANOS) {
+      return duration.get(unit);
+    }
+
+    return period.get(unit);
+  }
+
+  @Override
+  public List<TemporalUnit> getUnits() {
+    return units;
+  }
+
+  @Override
+  public Temporal addTo(Temporal temporal) {
+    return temporal.plus(period).plus(duration);
+  }
+
+  @Override
+  public Temporal subtractFrom(Temporal temporal) {
+    return temporal.minus(period).minus(duration);
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {
@@ -71,36 +130,36 @@ public String toString() {
     return period.toString() + duration.toString().substring(1);
   }
 
+  private boolean isCalendarBased() {
+    return !getPeriod().isZero() || getDuration().compareTo(ACCURATE_DURATION_UPPER_BOUND) >= 0;
+  }
+
   /**
    * Only supports a subset of ISO8601, combining both period and duration.
    *
    * @param text ISO8601 conforming interval expression
    * @return parsed interval
    */
   public static Interval parse(String text) {
-    final int timeOffset = text.lastIndexOf("T");
-    final Period period;
-    final Duration duration;
-
-    // to remain consistent with normal duration parsing which requires a duration to start with P
-    if (text.charAt(0) != 'P') {
-      throw new DateTimeParseException("Must start with P", text, 0);
-    }
-
-    if (timeOffset > 0) {
-      duration = Duration.parse(String.format("P%S", text.substring(timeOffset)));
-    } else {
-      duration = Duration.ZERO;
+    String sign = "";
+    int startOffset = 0;
+
+    if (text.startsWith("-")) {
+      startOffset = 1;
+      sign = "-";
+    } else if (text.startsWith("+")) {
+      startOffset = 1;
     }
 
-    if (timeOffset == -1) {
-      period = Period.parse(text);
-    } else if (timeOffset > 1) {
-      period = Period.parse(text.substring(0, timeOffset));
-    } else {
-      period = Period.ZERO;
+    final int durationOffset = text.indexOf('T');
+    if (durationOffset == -1) {
+      return new Interval(Period.parse(text), Duration.ZERO);
+    } else if (durationOffset == startOffset + 1) {
+      return new Interval(Period.ZERO, Duration.parse(text));
     }
 
-    return new Interval(period, duration);
+    return new Interval(
+        Period.parse(text.substring(0, durationOffset)),
+        Duration.parse(sign + "P" + text.substring(durationOffset)));
   }
 }

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

@@ -79,31 +79,25 @@ public static RepeatingInterval parse(String text) {
    * @return a RepeatingInterval based on the given text
    */
   public static RepeatingInterval parse(String text, String intervalDesignator) {
-    final int intervalDesignatorOffset = text.indexOf(intervalDesignator);
-    int repetitions = INFINITE;
-    final Interval interval;
-
-    if (text.charAt(0) != 'R') {
+    if (!text.startsWith("R")) {
       throw new DateTimeParseException("Repetition spec must start with R", text, 0);
     }
 
-    if (intervalDesignatorOffset == -1 || intervalDesignatorOffset == text.length() - 1) {
+    final int intervalDesignatorOffset = text.indexOf(intervalDesignator);
+    if (intervalDesignatorOffset == -1) {
       throw new DateTimeParseException("No interval given", text, intervalDesignatorOffset);
     }
 
-    final String intervalText = text.substring(intervalDesignatorOffset + 1);
-    interval = Interval.parse(intervalText);
-
-    if (intervalDesignatorOffset > 1) {
-      final String repetitionsText = text.substring(1, intervalDesignatorOffset);
-
-      try {
-        repetitions = Integer.parseInt(repetitionsText);
-      } catch (NumberFormatException e) {
-        throw new DateTimeParseException("Cannot parse repetitions count", repetitionsText, 1, e);
-      }
+    if (intervalDesignatorOffset == 1) { // startsWith("R/")
+      return new RepeatingInterval(INFINITE, Interval.parse(text.substring(2)));
     }
 
-    return new RepeatingInterval(repetitions, interval);
+    try {
+      return new RepeatingInterval(
+          Integer.parseInt(text.substring(1, intervalDesignatorOffset)),
+          Interval.parse(text.substring(intervalDesignatorOffset + 1)));
+    } catch (NumberFormatException e) {
+      throw new DateTimeParseException("Cannot parse repetitions count", text, 1, e);
+    }
   }
 }

bpmn-model/src/main/java/io/zeebe/model/bpmn/validation/zeebe/TimerEventDefinitionValidator.java

@@ -19,8 +19,8 @@
 import io.zeebe.model.bpmn.instance.TimeDate;
 import io.zeebe.model.bpmn.instance.TimeDuration;
 import io.zeebe.model.bpmn.instance.TimerEventDefinition;
+import io.zeebe.model.bpmn.util.time.Interval;
 import io.zeebe.model.bpmn.util.time.RepeatingInterval;
-import java.time.Duration;
 import java.time.format.DateTimeParseException;
 import org.camunda.bpm.model.xml.validation.ModelElementValidator;
 import org.camunda.bpm.model.xml.validation.ValidationResultCollector;
@@ -75,7 +75,7 @@ private void validateTimeCycle(
   private void validateTimeDuration(
       ValidationResultCollector validationResultCollector, TimeDuration timeDuration) {
     try {
-      Duration.parse(timeDuration.getTextContent());
+      Interval.parse(timeDuration.getTextContent());
     } catch (DateTimeParseException e) {
       validationResultCollector.addError(0, "Time duration is invalid");
     }

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

@@ -19,8 +19,14 @@
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.time.Duration;
+import java.time.LocalDateTime;
 import java.time.Period;
 import java.time.format.DateTimeParseException;
+import java.time.temporal.ChronoUnit;
+import java.time.temporal.UnsupportedTemporalTypeException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
 import org.junit.Test;
 
 public class IntervalTest {
@@ -66,6 +72,38 @@ public void shouldParseWithPeriodAndDuration() {
     assertThat(interval).isEqualTo(expected);
   }
 
+  @Test
+  public void shouldParseNegativeInterval() {
+    // given
+    final String text = "-P1Y2M4DT1H2M3S";
+    final Interval expected =
+        new Interval(
+            Period.of(-1, -2, -4),
+            Duration.ofHours(-1).plus(Duration.ofMinutes(-2)).plus(Duration.ofSeconds(-3)));
+
+    // when
+    final Interval interval = Interval.parse(text);
+
+    // then
+    assertThat(interval).isEqualTo(expected);
+  }
+
+  @Test
+  public void shouldParsePositiveInterval() {
+    // given
+    final String text = "+P1Y2M4DT1H2M3S";
+    final Interval expected =
+        new Interval(
+            Period.of(1, 2, 4),
+            Duration.ofHours(1).plus(Duration.ofMinutes(2)).plus(Duration.ofSeconds(3)));
+
+    // when
+    final Interval interval = Interval.parse(text);
+
+    // then
+    assertThat(interval).isEqualTo(expected);
+  }
+
   @Test
   public void shouldFailToParseWrongPeriod() {
     // given
@@ -101,4 +139,85 @@ public void shouldFailToParseIfNotStartingWithP() {
     // then
     assertThatThrownBy(() -> Interval.parse(text)).isInstanceOf(DateTimeParseException.class);
   }
+
+  @Test
+  public void shouldGetDurationTemporalUnit() {
+    // given
+    final Interval interval = new Interval(Period.of(1, 2, 3), Duration.ofSeconds(5, 35));
+    final ChronoUnit[] durationUnits = new ChronoUnit[] {ChronoUnit.SECONDS, ChronoUnit.NANOS};
+
+    // then
+    for (final ChronoUnit unit : durationUnits) {
+      assertThat(interval.get(unit)).isEqualTo(interval.getDuration().get(unit));
+    }
+  }
+
+  @Test
+  public void shouldGetPeriodTemporalUnit() {
+    // given
+    final Interval interval = new Interval(Period.of(1, 2, 3), Duration.ofSeconds(5, 35));
+    final ChronoUnit[] periodUnits =
+        new ChronoUnit[] {ChronoUnit.DAYS, ChronoUnit.MONTHS, ChronoUnit.YEARS};
+
+    // then
+    for (final ChronoUnit unit : periodUnits) {
+      assertThat(interval.get(unit)).isEqualTo(interval.getPeriod().get(unit));
+    }
+  }
+
+  @Test
+  public void shouldThrowExceptionOnGetUnsupportedTemporalUnit() {
+    // given
+    final Interval interval = new Interval(Period.of(1, 2, 3), Duration.ofSeconds(5, 35));
+    final List<ChronoUnit> supportedUnits =
+        Arrays.asList(
+            ChronoUnit.SECONDS,
+            ChronoUnit.NANOS,
+            ChronoUnit.DAYS,
+            ChronoUnit.MONTHS,
+            ChronoUnit.YEARS);
+    final List<ChronoUnit> unsupportedUnits =
+        Arrays.stream(ChronoUnit.values())
+            .filter(unit -> !supportedUnits.contains(unit))
+            .collect(Collectors.toList());
+
+    // then
+    for (final ChronoUnit unit : unsupportedUnits) {
+      assertThatThrownBy(() -> interval.get(unit))
+          .isInstanceOf(UnsupportedTemporalTypeException.class);
+    }
+  }
+
+  @Test
+  public void shouldReturnAllPeriodAndDurationUnits() {
+    // given
+    final Interval interval = new Interval(Period.of(1, 2, 3), Duration.ofSeconds(5, 35));
+
+    // then
+    assertThat(interval.getUnits())
+        .containsExactlyInAnyOrder(
+            ChronoUnit.SECONDS,
+            ChronoUnit.NANOS,
+            ChronoUnit.MONTHS,
+            ChronoUnit.DAYS,
+            ChronoUnit.YEARS);
+  }
+
+  @Test
+  public void shouldAddToTemporalAmount() {
+    // given
+    final Period period = Period.of(1, 2, 3);
+    final Duration duration = Duration.ofSeconds(5, 35);
+    final Interval interval = new Interval(period, duration);
+    final LocalDateTime amount = LocalDateTime.parse("2007-12-03T10:15:30");
+    final LocalDateTime expected = amount.plus(period).plus(duration);
+
+    // then
+    assertThat(interval.addTo(amount)).isEqualTo(expected);
+  }
+
+  @Test
+  public void shouldFailToParseEmptyString() {
+    assertThatThrownBy(() -> Interval.parse("")).isInstanceOf(DateTimeParseException.class);
+  }
 }

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

@@ -122,4 +122,9 @@ public void shouldFailToParseIfRepetitionCountDoesNotStartWithR() {
     assertThatThrownBy(() -> RepeatingInterval.parse(text))
         .isInstanceOf(DateTimeParseException.class);
   }
+
+  @Test
+  public void shouldFailToParseEmptyString() {
+    assertThatThrownBy(() -> Interval.parse("")).isInstanceOf(DateTimeParseException.class);
+  }
 }

bpmn-model/src/test/java/io/zeebe/model/bpmn/validation/ZeebeTimerValidationTest.java

@@ -38,6 +38,14 @@ public static Object[][] parameters() {
             .done(),
         singletonList(expect(TimerEventDefinition.class, "Time duration is invalid"))
       },
+      {
+        Bpmn.createExecutableProcess("process")
+            .startEvent()
+            .intermediateCatchEvent("catch", c -> c.timerWithDuration("R/PT01S"))
+            .endEvent()
+            .done(),
+        singletonList(expect(TimerEventDefinition.class, "Time duration is invalid"))
+      },
       {
         Bpmn.createExecutableProcess("process")
             .startEvent()