availability;
+ private boolean noMoreSplits;
+ private boolean closed;
+ private boolean splitAssigned;
+ private long lastEmittedNumber;
+ private long startTimestampSec;
+
+ public MetronomeReader(SourceReaderContext readerContext, @Nullable Long numberOfRows) {
+ this.readerContext = readerContext;
+ this.numberOfRows = numberOfRows;
+ this.availability = new CompletableFuture<>();
+ this.availabilityExecutor =
+ Executors.newSingleThreadScheduledExecutor(new ExecutorThreadFactory("metronome-source"));
+ this.lastEmittedNumber = 0L;
+ this.startTimestampSec = UNINITIALIZED;
+ }
+
+ @Override
+ public void start() {
+ readerContext.sendSplitRequest();
+ }
+
+ /**
+ * Emits the next due metronome row, or schedules the reader to become available at the next
+ * second boundary.
+ *
+ * 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.
+ */
+ @Override
+ public InputStatus pollNext(ReaderOutput output) {
+ if (closed) {
+ return InputStatus.END_OF_INPUT;
+ }
+
+ if (!splitAssigned) {
+ return noMoreSplits ? InputStatus.END_OF_INPUT : InputStatus.NOTHING_AVAILABLE;
+ }
+
+ long currentTimestampSec = currentTimestampSec();
+ if (startTimestampSec == UNINITIALIZED) {
+ startTimestampSec = currentTimestampSec;
+ }
+
+ long targetNumber = currentTimestampSec - startTimestampSec;
+ if (numberOfRows != null) {
+ targetNumber = Math.min(targetNumber, numberOfRows);
+ }
+
+ if (lastEmittedNumber < targetNumber) {
+ long nextNumber = lastEmittedNumber + 1;
+ long eventTimestampSec = startTimestampSec + nextNumber;
+ long eventTimestampMillis = eventTimestampSec * 1000L;
+ var row =
+ GenericRowData.of(
+ nextNumber, TimestampData.fromInstant(Instant.ofEpochSecond(eventTimestampSec)));
+ lastEmittedNumber = nextNumber;
+ output.collect(row, eventTimestampMillis);
+
+ return lastEmittedNumber < targetNumber ? InputStatus.MORE_AVAILABLE : nextStatus();
+ }
+
+ return nextStatus();
+ }
+
+ /**
+ * Snapshots the assigned split as the reader's progress state.
+ *
+ * The split carries the only source progress that must survive failover: the last emitted
+ * sequence number and the source's fixed start epoch second.
+ */
+ @Override
+ public List snapshotState(long checkpointId) {
+ if (!splitAssigned || closed || (numberOfRows != null && lastEmittedNumber >= numberOfRows)) {
+ return List.of();
+ }
+
+ return List.of(new MetronomeSplit(lastEmittedNumber, startTimestampSec));
+ }
+
+ /** Returns the current availability future used by the Flink runtime to avoid busy polling. */
+ @Override
+ public CompletableFuture isAvailable() {
+ return availability;
+ }
+
+ /** Restores progress from the assigned split and makes the reader immediately pollable. */
+ @Override
+ public void addSplits(List splits) {
+ for (var split : splits) {
+ lastEmittedNumber = split.lastEmittedNumber();
+ startTimestampSec = split.startTimestampSec();
+ splitAssigned = true;
+ break;
+ }
+ completeAvailability();
+ }
+
+ @Override
+ public void notifyNoMoreSplits() {
+ noMoreSplits = true;
+ completeAvailability();
+ }
+
+ /** Closes the reader and wakes any runtime caller blocked on the current availability future. */
+ @Override
+ public void close() {
+ closed = true;
+ availabilityExecutor.shutdownNow();
+ completeAvailability();
+ }
+
+ /**
+ * Determines the next source status after a poll cycle.
+ *
+ * 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.
+ */
+ private InputStatus nextStatus() {
+ if (lastEmittedNumber == Long.MAX_VALUE
+ || (numberOfRows != null && lastEmittedNumber >= numberOfRows)) {
+ closed = true;
+ return InputStatus.END_OF_INPUT;
+ }
+
+ if (noMoreSplits && startTimestampSec == UNINITIALIZED) {
+ return InputStatus.END_OF_INPUT;
+ }
+
+ availability = new CompletableFuture<>();
+ scheduleAvailability(nanosUntilNextSecond());
+
+ return InputStatus.NOTHING_AVAILABLE;
+ }
+
+ private void completeAvailability() {
+ availability.complete(null);
+ }
+
+ /** Schedules completion of the current availability future using nanosecond delay precision. */
+ private void scheduleAvailability(long delayNanos) {
+ if (!closed) {
+ availabilityExecutor.schedule(this::completeAvailability, delayNanos, TimeUnit.NANOSECONDS);
+ }
+ }
+
+ /** 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;
+ }
+}
diff --git a/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/MetronomeSource.java b/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/MetronomeSource.java
new file mode 100644
index 00000000..12fd55b2
--- /dev/null
+++ b/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/MetronomeSource.java
@@ -0,0 +1,81 @@
+/*
+ * Copyright © 2026 DataSQRL (contact@datasqrl.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datasqrl.flinkrunner.connector.datagen.metronome;
+
+import com.datasqrl.flinkrunner.connector.datagen.metronome.enumerator.MetronomeEnumerator;
+import com.datasqrl.flinkrunner.connector.datagen.metronome.enumerator.MetronomeEnumeratorState;
+import com.datasqrl.flinkrunner.connector.datagen.metronome.enumerator.MetronomeEnumeratorStateSerializer;
+import com.datasqrl.flinkrunner.connector.datagen.metronome.split.MetronomeSplit;
+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;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+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 {
+
+ @Serial private static final long serialVersionUID = 1L;
+
+ @Nullable private final Long numberOfRows;
+
+ public MetronomeSource() {
+ this(null);
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return numberOfRows == null ? Boundedness.CONTINUOUS_UNBOUNDED : Boundedness.BOUNDED;
+ }
+
+ @Override
+ public SourceReader createReader(SourceReaderContext readerContext) {
+ return new MetronomeReader(readerContext, numberOfRows);
+ }
+
+ @Override
+ public SplitEnumerator createEnumerator(
+ SplitEnumeratorContext enumContext) {
+
+ return new MetronomeEnumerator(enumContext, MetronomeEnumeratorState.unassigned());
+ }
+
+ @Override
+ public SplitEnumerator restoreEnumerator(
+ SplitEnumeratorContext enumContext,
+ MetronomeEnumeratorState checkpointState) {
+
+ return new MetronomeEnumerator(enumContext, checkpointState);
+ }
+
+ @Override
+ public SimpleVersionedSerializer getSplitSerializer() {
+ return new MetronomeSplitSerializer();
+ }
+
+ @Override
+ public SimpleVersionedSerializer getEnumeratorCheckpointSerializer() {
+ return new MetronomeEnumeratorStateSerializer();
+ }
+}
diff --git a/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/MetronomeTableSource.java b/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/MetronomeTableSource.java
new file mode 100644
index 00000000..caf42aeb
--- /dev/null
+++ b/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/MetronomeTableSource.java
@@ -0,0 +1,66 @@
+/*
+ * Copyright © 2026 DataSQRL (contact@datasqrl.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datasqrl.flinkrunner.connector.datagen.metronome;
+
+import javax.annotation.Nullable;
+import lombok.AccessLevel;
+import lombok.AllArgsConstructor;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.connector.source.SourceProvider;
+import org.apache.flink.table.connector.source.abilities.SupportsLimitPushDown;
+import org.apache.flink.table.types.DataType;
+
+/** Table source adapter that exposes the metronome source to Flink SQL. */
+@AllArgsConstructor(access = AccessLevel.PRIVATE)
+public class MetronomeTableSource implements ScanTableSource, SupportsLimitPushDown {
+
+ private static final int SOURCE_PARALLELISM = 1;
+
+ private final DataType rowDataType;
+
+ @Nullable private Long numberOfRows;
+
+ public MetronomeTableSource(DataType rowDataType) {
+ this(rowDataType, null);
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return ChangelogMode.insertOnly();
+ }
+
+ @Override
+ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {
+ return SourceProvider.of(new MetronomeSource(numberOfRows), SOURCE_PARALLELISM);
+ }
+
+ @Override
+ public DynamicTableSource copy() {
+ return new MetronomeTableSource(rowDataType, numberOfRows);
+ }
+
+ @Override
+ public String asSummaryString() {
+ return "MetronomeSource";
+ }
+
+ @Override
+ public void applyLimit(long limit) {
+ this.numberOfRows = limit;
+ }
+}
diff --git a/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/enumerator/MetronomeEnumerator.java b/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/enumerator/MetronomeEnumerator.java
new file mode 100644
index 00000000..f8b08a13
--- /dev/null
+++ b/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/enumerator/MetronomeEnumerator.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright © 2026 DataSQRL (contact@datasqrl.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datasqrl.flinkrunner.connector.datagen.metronome.enumerator;
+
+import com.datasqrl.flinkrunner.connector.datagen.metronome.split.MetronomeSplit;
+import java.util.List;
+import javax.annotation.Nullable;
+import lombok.AllArgsConstructor;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+
+/**
+ * Enumerator that owns the single metronome split and reassigns pending split state on recovery.
+ */
+@AllArgsConstructor
+public final class MetronomeEnumerator
+ implements SplitEnumerator {
+
+ private final SplitEnumeratorContext context;
+ private MetronomeEnumeratorState state;
+
+ @Override
+ public void start() {}
+
+ @Override
+ public void handleSplitRequest(int subtaskId, @Nullable String requesterHostname) {
+ if (state.getPendingSplit() != null) {
+ context.assignSplit(state.getPendingSplit(), subtaskId);
+ state = MetronomeEnumeratorState.assigned();
+
+ } else if (!state.isAssigned()) {
+ context.assignSplit(MetronomeSplit.initial(), subtaskId);
+ state = MetronomeEnumeratorState.assigned();
+
+ } else {
+ context.signalNoMoreSplits(subtaskId);
+ }
+ }
+
+ @Override
+ public void addSplitsBack(List splits, int subtaskId) {
+ if (!splits.isEmpty()) {
+ state = MetronomeEnumeratorState.pending(splits.get(0));
+ }
+ }
+
+ @Override
+ public void addReader(int subtaskId) {}
+
+ @Override
+ public MetronomeEnumeratorState snapshotState(long checkpointId) {
+ return state;
+ }
+
+ @Override
+ public void close() {}
+}
diff --git a/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/enumerator/MetronomeEnumeratorState.java b/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/enumerator/MetronomeEnumeratorState.java
new file mode 100644
index 00000000..ac26c002
--- /dev/null
+++ b/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/enumerator/MetronomeEnumeratorState.java
@@ -0,0 +1,46 @@
+/*
+ * Copyright © 2026 DataSQRL (contact@datasqrl.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datasqrl.flinkrunner.connector.datagen.metronome.enumerator;
+
+import com.datasqrl.flinkrunner.connector.datagen.metronome.split.MetronomeSplit;
+import javax.annotation.Nullable;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+/** Checkpointed enumerator state for initial assignment and pending split reassignment. */
+@Getter
+@RequiredArgsConstructor(access = lombok.AccessLevel.PRIVATE)
+public final class MetronomeEnumeratorState {
+
+ private final boolean assigned;
+ @Nullable private final MetronomeSplit pendingSplit;
+
+ public static MetronomeEnumeratorState of(boolean assigned) {
+ return assigned ? assigned() : unassigned();
+ }
+
+ public static MetronomeEnumeratorState assigned() {
+ return new MetronomeEnumeratorState(true, null);
+ }
+
+ public static MetronomeEnumeratorState unassigned() {
+ return new MetronomeEnumeratorState(false, null);
+ }
+
+ public static MetronomeEnumeratorState pending(MetronomeSplit split) {
+ return new MetronomeEnumeratorState(false, split);
+ }
+}
diff --git a/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/enumerator/MetronomeEnumeratorStateSerializer.java b/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/enumerator/MetronomeEnumeratorStateSerializer.java
new file mode 100644
index 00000000..46692037
--- /dev/null
+++ b/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/enumerator/MetronomeEnumeratorStateSerializer.java
@@ -0,0 +1,60 @@
+/*
+ * Copyright © 2026 DataSQRL (contact@datasqrl.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datasqrl.flinkrunner.connector.datagen.metronome.enumerator;
+
+import com.datasqrl.flinkrunner.connector.datagen.metronome.split.MetronomeSplit;
+import java.io.IOException;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataOutputSerializer;
+
+/** Serializer for enumerator checkpoint state. */
+public final class MetronomeEnumeratorStateSerializer
+ implements SimpleVersionedSerializer {
+
+ @Override
+ public int getVersion() {
+ return 1;
+ }
+
+ @Override
+ public byte[] serialize(MetronomeEnumeratorState state) throws IOException {
+ var out = new DataOutputSerializer(18);
+ 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();
+ }
+
+ @Override
+ public MetronomeEnumeratorState deserialize(int version, byte[] serialized) throws IOException {
+ var in = new DataInputDeserializer(serialized);
+
+ boolean assigned = in.readBoolean();
+ boolean hasPendingSplit = in.readBoolean();
+
+ return hasPendingSplit
+ ? MetronomeEnumeratorState.pending(new MetronomeSplit(in.readLong(), in.readLong()))
+ : MetronomeEnumeratorState.of(assigned);
+ }
+}
diff --git a/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/split/MetronomeSplit.java b/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/split/MetronomeSplit.java
new file mode 100644
index 00000000..74fc91ce
--- /dev/null
+++ b/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/split/MetronomeSplit.java
@@ -0,0 +1,33 @@
+/*
+ * Copyright © 2026 DataSQRL (contact@datasqrl.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+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 static MetronomeSplit initial() {
+ return new MetronomeSplit(0L, MetronomeReader.UNINITIALIZED);
+ }
+
+ @Override
+ public String splitId() {
+ return "metronome";
+ }
+}
diff --git a/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/split/MetronomeSplitSerializer.java b/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/split/MetronomeSplitSerializer.java
new file mode 100644
index 00000000..2f6795ff
--- /dev/null
+++ b/connectors/datagen-connectors/src/main/java/com/datasqrl/flinkrunner/connector/datagen/metronome/split/MetronomeSplitSerializer.java
@@ -0,0 +1,46 @@
+/*
+ * Copyright © 2026 DataSQRL (contact@datasqrl.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datasqrl.flinkrunner.connector.datagen.metronome.split;
+
+import java.io.IOException;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataOutputSerializer;
+
+/** Serializer for the metronome split progress state. */
+public final class MetronomeSplitSerializer implements SimpleVersionedSerializer {
+
+ @Override
+ public int getVersion() {
+ return 1;
+ }
+
+ @Override
+ public byte[] serialize(MetronomeSplit split) throws IOException {
+ var out = new DataOutputSerializer(16);
+ out.writeLong(split.lastEmittedNumber());
+ out.writeLong(split.startTimestampSec());
+
+ return out.getCopyOfBuffer();
+ }
+
+ @Override
+ public MetronomeSplit deserialize(int version, byte[] serialized) throws IOException {
+ var in = new DataInputDeserializer(serialized);
+
+ return new MetronomeSplit(in.readLong(), in.readLong());
+ }
+}
diff --git a/connectors/datagen-connectors/src/test/java/com/datasqrl/flinkrunner/connector/datagen/metronome/MetronomeSourceIT.java b/connectors/datagen-connectors/src/test/java/com/datasqrl/flinkrunner/connector/datagen/metronome/MetronomeSourceIT.java
new file mode 100644
index 00000000..35abc4a2
--- /dev/null
+++ b/connectors/datagen-connectors/src/test/java/com/datasqrl/flinkrunner/connector/datagen/metronome/MetronomeSourceIT.java
@@ -0,0 +1,173 @@
+/*
+ * Copyright © 2026 DataSQRL (contact@datasqrl.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datasqrl.flinkrunner.connector.datagen.metronome;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.datasqrl.flinkrunner.connector.datagen.metronome.split.MetronomeSplit;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.flink.api.common.eventtime.Watermark;
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.SourceEvent;
+import org.apache.flink.api.connector.source.SourceOutput;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.metrics.groups.SourceReaderMetricGroup;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.table.api.internal.TableEnvironmentImpl;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.UserCodeClassLoader;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/** Integration tests for the metronome SQL connector. */
+@ExtendWith(MiniClusterExtension.class)
+class MetronomeSourceIT { // extends TableITCaseBase {
+
+ private TableEnvironment tEnv;
+
+ @BeforeEach
+ void beforeEach() {
+ tEnv = TableEnvironmentImpl.create(EnvironmentSettings.newInstance().inStreamingMode().build());
+ tEnv.getConfig().set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 1);
+ }
+
+ @Test
+ @Timeout(30)
+ void happyPathEmitsMonotonicSequence() throws Exception {
+ tEnv.executeSql(
+ """
+ CREATE TABLE metronome_source (
+ num BIGINT,
+ ts TIMESTAMP_LTZ(3),
+ WATERMARK FOR ts AS ts
+ ) WITH (
+ 'connector' = 'metronome'
+ )""");
+
+ assertThat(collectRows("SELECT num FROM metronome_source", 3))
+ .containsExactly(Row.of(1L), Row.of(2L), Row.of(3L));
+ }
+
+ @Test
+ void restoresReaderProgressFromCheckpointedSplit() {
+ 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)));
+
+ assertThat(reader.pollNext(output)).isEqualTo(InputStatus.MORE_AVAILABLE);
+ assertThat(output.numbers()).containsExactly(2L);
+ assertThat(reader.snapshotState(1L))
+ .containsExactly(new MetronomeSplit(2L, startTimestampSec));
+ } finally {
+ reader.close();
+ }
+ }
+
+ private List collectRows(String sql, int rowCount) throws Exception {
+ List rows = new ArrayList<>();
+ try (CloseableIterator iterator = tEnv.executeSql(sql).collect()) {
+ while (rows.size() < rowCount) {
+ rows.add(iterator.next());
+ }
+ }
+ return rows;
+ }
+
+ private static SourceReaderContext unusedReaderContext() {
+ return new SourceReaderContext() {
+ @Override
+ public SourceReaderMetricGroup metricGroup() {
+ return null;
+ }
+
+ @Override
+ public Configuration getConfiguration() {
+ return new Configuration();
+ }
+
+ @Override
+ public String getLocalHostName() {
+ return "localhost";
+ }
+
+ @Override
+ public int getIndexOfSubtask() {
+ return 0;
+ }
+
+ @Override
+ public void sendSplitRequest() {}
+
+ @Override
+ public void sendSourceEventToCoordinator(SourceEvent sourceEvent) {}
+
+ @Override
+ public UserCodeClassLoader getUserCodeClassLoader() {
+ return null;
+ }
+ };
+ }
+
+ private static final class CollectingReaderOutput implements ReaderOutput {
+
+ private final List rows = new ArrayList<>();
+
+ @Override
+ public void collect(RowData record) {
+ rows.add(record);
+ }
+
+ @Override
+ public void collect(RowData record, long timestamp) {
+ rows.add(record);
+ }
+
+ @Override
+ public void emitWatermark(Watermark watermark) {}
+
+ @Override
+ public void markIdle() {}
+
+ @Override
+ public void markActive() {}
+
+ @Override
+ public SourceOutput createOutputForSplit(String splitId) {
+ return this;
+ }
+
+ @Override
+ public void releaseOutputForSplit(String splitId) {}
+
+ private List numbers() {
+ return rows.stream().map(row -> row.getLong(0)).toList();
+ }
+ }
+}
diff --git a/connectors/pom.xml b/connectors/pom.xml
index aba84ea8..23243fb4 100644
--- a/connectors/pom.xml
+++ b/connectors/pom.xml
@@ -30,6 +30,7 @@
Connectors
+ datagen-connectors
kafka-safe-connector
postgresql-connector
diff --git a/flink-sql-runner/pom.xml b/flink-sql-runner/pom.xml
index 2695478c..9f2beb78 100644
--- a/flink-sql-runner/pom.xml
+++ b/flink-sql-runner/pom.xml
@@ -127,6 +127,12 @@
+
+ com.datasqrl.flinkrunner
+ datagen-connectors
+ ${project.version}
+
+
com.datasqrl.flinkrunner
kafka-safe-connector