feat: Use current timestamp when metronome replays or catches up (#332)

Author: ferenc-csaky

connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/MetronomeReader.java

@@ -68,8 +68,9 @@ public void start() {
    * second boundary.
    *
    * <p>The emitted sequence number is derived from wall-clock epoch seconds relative to the first
-   * observed start second. If the reader wakes up late, repeated calls emit all missing sequence
-   * numbers with their intended event timestamps until the source catches up.
+   * observed start second. If the reader wakes up late or recovers from a checkpoint, repeated
+   * calls emit all missing sequence numbers with the current wall-clock timestamp until the source
+   * catches up.
    */
   @Override
   public InputStatus pollNext(ReaderOutput<RowData> output) {
@@ -93,11 +94,10 @@ public InputStatus pollNext(ReaderOutput<RowData> output) {
 
     if (lastEmittedNumber < targetNumber) {
       long nextNumber = lastEmittedNumber + 1;
-      long eventTimestampSec = startTimestampSec + nextNumber;
-      long eventTimestampMillis = eventTimestampSec * 1000L;
+      long eventTimestampMillis = currentTimestampSec * 1000L;
       var row =
           GenericRowData.of(
-              nextNumber, TimestampData.fromInstant(Instant.ofEpochSecond(eventTimestampSec)));
+              nextNumber, TimestampData.fromInstant(Instant.ofEpochSecond(currentTimestampSec)));
       lastEmittedNumber = nextNumber;
       output.collect(row, eventTimestampMillis);
 

connectors/datagen-connectors/src/test/java/com/datasqrl/flinkrunner/connector/datagen/metronome/MetronomeSourceIT.java

@@ -73,16 +73,21 @@ ts TIMESTAMP_LTZ(3),
   }
 
   @Test
-  void restoresReaderProgressFromCheckpointedSplit() {
+  void restoresReaderProgressWithCurrentTimestampFromCheckpointedSplit() {
     long startTimestampSec = Instant.now().getEpochSecond() - 3L;
     var reader = new MetronomeReader(unusedReaderContext(), 3L);
     var output = new CollectingReaderOutput();
 
     try {
       reader.addSplits(List.of(new MetronomeSplit(1L, startTimestampSec)));
 
+      long beforePollSec = Instant.now().getEpochSecond();
       assertThat(reader.pollNext(output)).isEqualTo(InputStatus.MORE_AVAILABLE);
+      long afterPollSec = Instant.now().getEpochSecond();
+
       assertThat(output.numbers()).containsExactly(2L);
+      assertThat(output.eventTimestampSecs().get(0)).isBetween(beforePollSec, afterPollSec);
+      assertThat(output.rowTimestampSecs().get(0)).isBetween(beforePollSec, afterPollSec);
       assertThat(reader.snapshotState(1L))
           .containsExactly(new MetronomeSplit(2L, startTimestampSec));
     } finally {
@@ -138,6 +143,7 @@ public UserCodeClassLoader getUserCodeClassLoader() {
   private static final class CollectingReaderOutput implements ReaderOutput<RowData> {
 
     private final List<RowData> rows = new ArrayList<>();
+    private final List<Long> eventTimestamps = new ArrayList<>();
 
     @Override
     public void collect(RowData record) {
@@ -147,6 +153,7 @@ public void collect(RowData record) {
     @Override
     public void collect(RowData record, long timestamp) {
       rows.add(record);
+      eventTimestamps.add(timestamp);
     }
 
     @Override
@@ -169,5 +176,13 @@ public void releaseOutputForSplit(String splitId) {}
     private List<Long> numbers() {
       return rows.stream().map(row -> row.getLong(0)).toList();
     }
+
+    private List<Long> eventTimestampSecs() {
+      return eventTimestamps.stream().map(timestamp -> timestamp / 1000L).toList();
+    }
+
+    private List<Long> rowTimestampSecs() {
+      return rows.stream().map(row -> row.getTimestamp(1, 3).toInstant().getEpochSecond()).toList();
+    }
   }
 }