feat: Rework metronome source to emit AT MOST one event per sec (#338)

Author: ferenc-csaky

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

@@ -18,7 +18,6 @@
 import com.google.auto.service.AutoService;
 import java.util.Set;
 import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.configuration.ConfigOptions;
 import org.apache.flink.table.api.ValidationException;
 import org.apache.flink.table.connector.source.DynamicTableSource;
 import org.apache.flink.table.factories.DynamicTableSourceFactory;
@@ -33,13 +32,6 @@ public class MetronomeDynamicTableSourceFactory implements DynamicTableSourceFac
 
   public static final String IDENTIFIER = "metronome";
 
-  public static final ConfigOption<Boolean> REPLAY_ON_FAILURE =
-      ConfigOptions.key("replay-on-failure")
-          .booleanType()
-          .defaultValue(true)
-          .withDescription(
-              "Whether to emit sequence numbers missed while recovering from a checkpoint.");
-
   @Override
   public DynamicTableSource createDynamicTableSource(Context context) {
     var helper = FactoryUtil.createTableFactoryHelper(this, context);
@@ -48,7 +40,7 @@ public DynamicTableSource createDynamicTableSource(Context context) {
     var rowDataType = context.getPhysicalRowDataType();
     validateSchema(rowDataType);
 
-    return new MetronomeTableSource(rowDataType, helper.getOptions().get(REPLAY_ON_FAILURE));
+    return new MetronomeTableSource(rowDataType);
   }
 
   @Override
@@ -63,7 +55,7 @@ public Set<ConfigOption<?>> requiredOptions() {
 
   @Override
   public Set<ConfigOption<?>> optionalOptions() {
-    return Set.of(REPLAY_ON_FAILURE);
+    return Set.of();
   }
 
   private static void validateSchema(DataType rowDataType) {

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

@@ -38,32 +38,24 @@ public class MetronomeReader implements SourceReader<RowData, MetronomeSplit> {
   public static final long UNINITIALIZED = Long.MIN_VALUE;
 
   private final SourceReaderContext readerContext;
-  private final boolean replayOnFailure;
   @Nullable private final Long numberOfRows;
   private final ScheduledExecutorService availabilityExecutor;
 
   private CompletableFuture<Void> availability;
   private boolean noMoreSplits;
   private boolean closed;
   private boolean splitAssigned;
-  private boolean skipReplayOnRecovery;
   private long lastEmittedNumber;
-  private long startTimestampSec;
+  private long lastEmissionTimestampMillis;
 
-  public MetronomeReader(
-      SourceReaderContext readerContext, boolean replayOnFailure, @Nullable Long numberOfRows) {
+  public MetronomeReader(SourceReaderContext readerContext, @Nullable Long numberOfRows) {
     this.readerContext = readerContext;
     this.numberOfRows = numberOfRows;
-    this.replayOnFailure = replayOnFailure;
     this.availability = new CompletableFuture<>();
     this.availabilityExecutor =
         Executors.newSingleThreadScheduledExecutor(new ExecutorThreadFactory("metronome-source"));
     this.lastEmittedNumber = 0L;
-    this.startTimestampSec = UNINITIALIZED;
-  }
-
-  public MetronomeReader(SourceReaderContext readerContext, @Nullable Long numberOfRows) {
-    this(readerContext, true, numberOfRows);
+    this.lastEmissionTimestampMillis = UNINITIALIZED;
   }
 
   @Override
@@ -75,11 +67,9 @@ public void start() {
    * Emits the next due metronome row, or schedules the reader to become available at the next
    * 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 the current wall-clock timestamp until the source catches up. If checkpoint
-   * recovery replay is disabled, only the first poll after restoring checkpointed progress skips
-   * the recovery backlog.
+   * <p>The emitted sequence number is always the next number after the checkpointed progress. The
+   * reader never drains wall-clock backlog in a burst: after every emitted row, it records the
+   * current time so the next row cannot become due before a full second has elapsed.
    */
   @Override
   public InputStatus pollNext(ReaderOutput<RowData> output) {
@@ -91,37 +81,16 @@ public InputStatus pollNext(ReaderOutput<RowData> output) {
       return noMoreSplits ? InputStatus.END_OF_INPUT : InputStatus.NOTHING_AVAILABLE;
     }
 
-    long currentTimestampSec = currentTimestampSec();
-    if (startTimestampSec == UNINITIALIZED) {
-      startTimestampSec = currentTimestampSec;
-    }
+    var currentTimestamp = Instant.now();
+    long currentTimestampMillis = currentTimestamp.toEpochMilli();
 
-    long targetNumber = currentTimestampSec - startTimestampSec;
-    if (numberOfRows != null) {
-      targetNumber = Math.min(targetNumber, numberOfRows);
-    }
-
-    // One-shot checkpoint recovery skip; normal late wakeups still replay.
-    boolean skipReplay = skipReplayOnRecovery;
-    skipReplayOnRecovery = false;
-
-    if (lastEmittedNumber < targetNumber) {
+    if (isDue(currentTimestampMillis)
+        && (numberOfRows == null || lastEmittedNumber < numberOfRows)) {
       long nextNumber = lastEmittedNumber + 1;
-      long eventTimestampMillis = currentTimestampSec * 1000L;
-      var row =
-          GenericRowData.of(
-              nextNumber, TimestampData.fromInstant(Instant.ofEpochSecond(currentTimestampSec)));
+      var row = GenericRowData.of(nextNumber, TimestampData.fromInstant(currentTimestamp));
       lastEmittedNumber = nextNumber;
-      if (skipReplay) {
-        // Realign time so the restored backlog is dropped after this record.
-        startTimestampSec = currentTimestampSec - lastEmittedNumber;
-      }
-      output.collect(row, eventTimestampMillis);
-
-      // Skip waits for the next tick; otherwise drain due records until caught up.
-      return !skipReplay && lastEmittedNumber < targetNumber
-          ? InputStatus.MORE_AVAILABLE
-          : nextStatus();
+      lastEmissionTimestampMillis = currentTimestampMillis;
+      output.collect(row, currentTimestampMillis);
     }
 
     return nextStatus();
@@ -131,15 +100,15 @@ public InputStatus pollNext(ReaderOutput<RowData> output) {
    * Snapshots the assigned split as the reader's progress state.
    *
    * <p>The split carries the only source progress that must survive failover: the last emitted
-   * sequence number and the source's effective start epoch second.
+   * sequence number.
    */
   @Override
   public List<MetronomeSplit> snapshotState(long checkpointId) {
     if (!splitAssigned || closed || (numberOfRows != null && lastEmittedNumber >= numberOfRows)) {
       return List.of();
     }
 
-    return List.of(new MetronomeSplit(lastEmittedNumber, startTimestampSec));
+    return List.of(new MetronomeSplit(lastEmittedNumber));
   }
 
   /** Returns the current availability future used by the Flink runtime to avoid busy polling. */
@@ -153,10 +122,6 @@ public CompletableFuture<Void> isAvailable() {
   public void addSplits(List<MetronomeSplit> splits) {
     for (var split : splits) {
       lastEmittedNumber = split.lastEmittedNumber();
-      startTimestampSec = split.startTimestampSec();
-      // Non-initial split means checkpoint/savepoint recovery.
-      skipReplayOnRecovery =
-          !replayOnFailure && (startTimestampSec != UNINITIALIZED || lastEmittedNumber > 0L);
       splitAssigned = true;
       break;
     }
@@ -181,8 +146,8 @@ public void close() {
    * Determines the next source status after a poll cycle.
    *
    * <p>The source ends when bounded output is exhausted or the sequence reaches {@link
-   * Long#MAX_VALUE}. Otherwise, it replaces the availability future and schedules it for the next
-   * epoch-second boundary.
+   * Long#MAX_VALUE}. Otherwise, it replaces the availability future and schedules it one second
+   * later.
    */
   private InputStatus nextStatus() {
     if (lastEmittedNumber == Long.MAX_VALUE
@@ -191,36 +156,28 @@ private InputStatus nextStatus() {
       return InputStatus.END_OF_INPUT;
     }
 
-    if (noMoreSplits && startTimestampSec == UNINITIALIZED) {
-      return InputStatus.END_OF_INPUT;
-    }
-
     availability = new CompletableFuture<>();
-    scheduleAvailability(nanosUntilNextSecond());
+    scheduleAvailability();
 
     return InputStatus.NOTHING_AVAILABLE;
   }
 
+  private boolean isDue(long currentTimestampMillis) {
+    return lastEmissionTimestampMillis == UNINITIALIZED
+        || currentTimestampMillis - lastEmissionTimestampMillis >= TimeUnit.SECONDS.toMillis(1);
+  }
+
   private void completeAvailability() {
     availability.complete(null);
   }
 
-  /** Schedules completion of the current availability future using nanosecond delay precision. */
-  private void scheduleAvailability(long delayNanos) {
+  /** Schedules completion of the current availability future one second later. */
+  private void scheduleAvailability() {
     if (!closed) {
-      availabilityExecutor.schedule(this::completeAvailability, delayNanos, TimeUnit.NANOSECONDS);
+      // Capture the current future so stale timers cannot wake newer poll cycles
+      var scheduledAvailability = availability;
+      availabilityExecutor.schedule(
+          () -> scheduledAvailability.complete(null), 1, TimeUnit.SECONDS);
     }
   }
-
-  /** Returns the current wall-clock epoch second used for sequence catch-up calculations. */
-  private static long currentTimestampSec() {
-    return Instant.now().getEpochSecond();
-  }
-
-  /** Returns the remaining nanoseconds until the next wall-clock epoch-second boundary. */
-  private static long nanosUntilNextSecond() {
-    int currentNano = Instant.now().getNano();
-
-    return TimeUnit.SECONDS.toNanos(1L) - currentNano;
-  }
 }

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

@@ -22,7 +22,6 @@
 import com.datasqrl.flinkrunner.connector.datagen.metronome.split.MetronomeSplitSerializer;
 import java.io.Serial;
 import javax.annotation.Nullable;
-import lombok.AllArgsConstructor;
 import org.apache.flink.api.connector.source.Boundedness;
 import org.apache.flink.api.connector.source.Source;
 import org.apache.flink.api.connector.source.SourceReader;
@@ -33,20 +32,18 @@
 import org.apache.flink.table.data.RowData;
 
 /** Source that produces a single global metronome sequence with event-time timestamps. */
-@AllArgsConstructor
 public class MetronomeSource implements Source<RowData, MetronomeSplit, MetronomeEnumeratorState> {
 
   @Serial private static final long serialVersionUID = 1L;
 
-  private final boolean replayOnFailure;
   @Nullable private final Long numberOfRows;
 
   public MetronomeSource() {
-    this(true, null);
+    this(null);
   }
 
   public MetronomeSource(@Nullable Long numberOfRows) {
-    this(true, numberOfRows);
+    this.numberOfRows = numberOfRows;
   }
 
   @Override
@@ -56,7 +53,7 @@ public Boundedness getBoundedness() {
 
   @Override
   public SourceReader<RowData, MetronomeSplit> createReader(SourceReaderContext readerContext) {
-    return new MetronomeReader(readerContext, replayOnFailure, numberOfRows);
+    return new MetronomeReader(readerContext, numberOfRows);
   }
 
   @Override

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

@@ -32,16 +32,11 @@ public class MetronomeTableSource implements ScanTableSource, SupportsLimitPushD
   private static final int SOURCE_PARALLELISM = 1;
 
   private final DataType rowDataType;
-  private final boolean replayOnFailure;
 
   @Nullable private Long numberOfRows;
 
   public MetronomeTableSource(DataType rowDataType) {
-    this(rowDataType, true, null);
-  }
-
-  public MetronomeTableSource(DataType rowDataType, boolean replayOnFailure) {
-    this(rowDataType, replayOnFailure, null);
+    this(rowDataType, null);
   }
 
   @Override
@@ -51,13 +46,12 @@ public ChangelogMode getChangelogMode() {
 
   @Override
   public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {
-    return SourceProvider.of(
-        new MetronomeSource(replayOnFailure, numberOfRows), SOURCE_PARALLELISM);
+    return SourceProvider.of(new MetronomeSource(numberOfRows), SOURCE_PARALLELISM);
   }
 
   @Override
   public DynamicTableSource copy() {
-    return new MetronomeTableSource(rowDataType, replayOnFailure, numberOfRows);
+    return new MetronomeTableSource(rowDataType, numberOfRows);
   }
 
   @Override

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

@@ -32,15 +32,14 @@ public int getVersion() {
 
   @Override
   public byte[] serialize(MetronomeEnumeratorState state) throws IOException {
-    var out = new DataOutputSerializer(18);
+    var out = new DataOutputSerializer(10);
     var pendingSplit = state.getPendingSplit();
 
     out.writeBoolean(state.isAssigned());
     out.writeBoolean(pendingSplit != null);
 
     if (pendingSplit != null) {
       out.writeLong(pendingSplit.lastEmittedNumber());
-      out.writeLong(pendingSplit.startTimestampSec());
     }
 
     return out.getCopyOfBuffer();
@@ -54,7 +53,7 @@ public MetronomeEnumeratorState deserialize(int version, byte[] serialized) thro
     boolean hasPendingSplit = in.readBoolean();
 
     return hasPendingSplit
-        ? MetronomeEnumeratorState.pending(new MetronomeSplit(in.readLong(), in.readLong()))
+        ? MetronomeEnumeratorState.pending(new MetronomeSplit(in.readLong()))
         : MetronomeEnumeratorState.of(assigned);
   }
 }

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

@@ -15,15 +15,13 @@
  */
 package com.datasqrl.flinkrunner.connector.datagen.metronome.split;
 
-import com.datasqrl.flinkrunner.connector.datagen.metronome.MetronomeReader;
 import org.apache.flink.api.connector.source.SourceSplit;
 
 /** Single source split carrying metronome reader progress across checkpoints and recovery. */
-public record MetronomeSplit(long lastEmittedNumber, long startTimestampSec)
-    implements SourceSplit {
+public record MetronomeSplit(long lastEmittedNumber) implements SourceSplit {
 
   public static MetronomeSplit initial() {
-    return new MetronomeSplit(0L, MetronomeReader.UNINITIALIZED);
+    return new MetronomeSplit(0L);
   }
 
   @Override

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

@@ -30,9 +30,8 @@ public int getVersion() {
 
   @Override
   public byte[] serialize(MetronomeSplit split) throws IOException {
-    var out = new DataOutputSerializer(16);
+    var out = new DataOutputSerializer(8);
     out.writeLong(split.lastEmittedNumber());
-    out.writeLong(split.startTimestampSec());
 
     return out.getCopyOfBuffer();
   }
@@ -41,6 +40,6 @@ public byte[] serialize(MetronomeSplit split) throws IOException {
   public MetronomeSplit deserialize(int version, byte[] serialized) throws IOException {
     var in = new DataInputDeserializer(serialized);
 
-    return new MetronomeSplit(in.readLong(), in.readLong());
+    return new MetronomeSplit(in.readLong());
   }
 }