StoatFlow

Scheduled sources

Generate records on an interval or a cron schedule with StreamsBuilder.scheduled() — a topology source that emits without consuming from Kafka.

StreamsBuilder.scheduled() is a StoatFlow extension — it has no Kafka Streams equivalent. It creates a topology source that periodically generates records and feeds them downstream, without consuming from any Kafka topic. Use it for heartbeats, periodic state-store scans, external polling, or any time-driven event you want to flow through the same DAG as your Kafka-sourced records.

Interval-based sources

The interval overload fires every Duration according to a TimeNotion:

  • WALL_CLOCK_TIME — fires on the system clock, every interval, regardless of data flow. Use for heartbeats, metrics, or polling that must happen on real-world time.
  • STREAM_TIME — fires when the event-time watermark advances past the interval. Data-driven: with no records flowing, the watermark doesn't move and the source doesn't fire.

The interval must be positive (scheduled() throws IllegalArgumentException otherwise).

import io.stoatflow.core.processor.PunctuationType
import io.stoatflow.core.topology.Produced
import io.stoatflow.core.topology.StreamsBuilder
import org.apache.kafka.common.serialization.Serdes
import java.time.Duration

val builder = StreamsBuilder()

// Emit a heartbeat every 30 seconds on wall-clock time
builder
    .scheduled<String, String>(
        interval = Duration.ofSeconds(30),
        type = PunctuationType.WALL_CLOCK_TIME,
        emitter = { context ->
            context.forward("heartbeat", "alive-${context.currentWallClockTime()}")
        },
    )
    .to("heartbeats", Produced.with(Serdes.String(), Serdes.String()))
For Java, the emitter is delivered as a java.util.function.Consumer<ScheduledEmitterContext<K, V>>, so a plain lambda works. The Kotlin lambda overload is hidden from Java (@JvmSynthetic) to avoid ambiguity. The named and keySerde parameters have Kotlin defaults; from Java, pass null to skip them.

Naming the source

Pass a Named to give the source a stable name in the topology graph and metrics. Without it, sources are auto-named scheduled-source-0, scheduled-source-1, … in declaration order — which is the name you'll reference when testing.

import io.stoatflow.core.processor.PunctuationType
import io.stoatflow.core.topology.Named
import java.time.Duration

builder.scheduled<String, String>(
    named = Named.`as`("heartbeat-source"),
    interval = Duration.ofSeconds(30),
    type = PunctuationType.WALL_CLOCK_TIME,
    emitter = { context -> context.forward("heartbeat", "alive") },
)

Cron-based sources

The cron overload fires according to a CronExpression. Cron scheduling always uses wall-clock time — there is no TimeNotion parameter, and cron sources do not fire on watermark advance.

import io.stoatflow.core.processor.CronExpression
import io.stoatflow.core.topology.Named
import io.stoatflow.core.topology.Produced
import org.apache.kafka.common.serialization.Serdes

// Fire every day at midnight (Unix 5-field cron)
builder
    .scheduled<String, String>(
        named = Named.`as`("daily-report-source"),
        cron = CronExpression.unix("0 0 * * *"),
        emitter = { context ->
            context.forward("report", "daily-${context.currentWallClockTime()}")
        },
    )
    .to("daily-reports", Produced.with(Serdes.String(), Serdes.String()))

Cron expressions

CronExpression wraps three cron dialects. Pick the factory that matches the dialect of your expression — the field counts differ, so an expression valid in one dialect is usually invalid in another. An unparseable expression throws IllegalArgumentException.

FactoryFieldsFormatExampleMeaning
CronExpression.unix(...)5min hour day-of-month month day-of-week"0 * * * *"top of every hour
CronExpression.quartz(...)6–7sec min hour day-of-month month day-of-week [year]"0 0 0 * * ?"every day at midnight
CronExpression.spring(...)6sec min hour day-of-month month day-of-week"0 0 0 * * *"every day at midnight

All three factories are @JvmStatic, so they're called identically from Kotlin and Java. Day-of-week numbering differs between dialects (Unix 0–6 with 0 = Sunday; Quartz 1–7 with 1 = Sunday), so consult the dialect when porting expressions.

CronExpression resolves the next fire time in the system default time zone by default. The next-execution helpers (nextExecutionTime, timeUntilNextExecution) accept an explicit ZoneId if you need a fixed zone.

The emitter and its context

The emitter is a ScheduledEmitter<K, V> — a single-method functional interface. Each invocation receives a ScheduledEmitterContext<K, V> for emitting records and reading state.

Forwarding records

The context offers three forward overloads:

MethodTimestampHeaders
forward(key, value)current wall-clock timenone
forward(key, value, timestamp)explicitnone
forward(record: Record<K, V>)from the Recordfrom the Record

A single emitter invocation may forward zero, one, or many records. The key drives downstream routing, exactly as for Kafka-sourced records, so per-key ordering downstream behaves the same way — see Lanes & parallelism.

import io.stoatflow.core.processor.Record
import org.apache.kafka.common.header.internals.RecordHeaders

builder.scheduled<String, String>(
    interval = Duration.ofSeconds(10),
    type = PunctuationType.WALL_CLOCK_TIME,
    emitter = { context ->
        // simplest: key + value, timestamped now
        context.forward("k1", "v1")

        // explicit event-time timestamp
        context.forward("k2", "v2", context.currentWallClockTime())

        // full control via Record (key, value, timestamp, headers)
        val headers = RecordHeaders().apply { add("source", "scheduled".toByteArray()) }
        context.forward(Record("k3", "v3", context.currentWallClockTime(), headers))
    },
)

Reading state stores

The context exposes getStateStore(name) for read-only access. Write operations (put, delete) on a store obtained this way throw UnsupportedOperationException. This makes the scheduled source a natural fit for periodically scanning state and emitting derived events — for example, scanning pending items and emitting a "publish" event once their scheduled time has passed.

import io.stoatflow.core.state.ReadOnlyKeyValueStore

builder.scheduled<String, String>(
    named = Named.`as`("publish-scanner"),
    interval = Duration.ofMinutes(1),
    type = PunctuationType.WALL_CLOCK_TIME,
    emitter = { context ->
        val now = context.currentWallClockTime()
        val pending = context.getStateStore<ReadOnlyKeyValueStore<String, Long>>("pending-publishes")
        pending.all().use { it ->
            it.forEach { entry ->
                if (entry.value <= now) {
                    context.forward(entry.key, "published")
                }
            }
        }
    },
)

all(), range(), prefixScan(), and the reverse variants all return a KeyValueIterator<K, V>, which is Closeable — close it (Kotlin use {}, Java try-with-resources) to release resources. See State stores for the full read-only query surface.

For the time helpers: currentWallClockTime() and currentSystemTimeMs() return system time; currentWatermarkMs() (also currentStreamTimeMs()) returns the global event-time watermark. For a STREAM_TIME source, currentWatermarkMs() is the watermark value that triggered the firing; for a WALL_CLOCK_TIME source it's whatever the global watermark currently is (Long.MIN_VALUE if no events have been processed yet).

Processing the result

scheduled() returns a KStream<K, V> — there's nothing special about it downstream. Map, filter, group, aggregate, join, or merge it with a Kafka-sourced stream, all with the operators from KStream & KTable.

val ticks = builder.scheduled<String, Int>(
    interval = Duration.ofSeconds(5),
    type = PunctuationType.WALL_CLOCK_TIME,
    emitter = { context ->
        context.forward("a", 1)
        context.forward("b", 2)
    },
)

ticks
    .filter { _, value -> value > 1 }
    .mapValues { value -> value * 10 }
    .to("output", Produced.with(Serdes.String(), Serdes.Integer()))

To combine a scheduled source with a Kafka topic, keep a reference to each KStream and merge them:

import io.stoatflow.core.topology.Consumed

val scheduled = builder.scheduled<String, String>(
    interval = Duration.ofSeconds(5),
    type = PunctuationType.WALL_CLOCK_TIME,
    emitter = { context -> context.forward("scheduled", "tick") },
)
val kafka = builder.stream<String, String>("input", Consumed.with(Serdes.String(), Serdes.String()))

scheduled
    .merge(kafka)
    .to("output", Produced.with(Serdes.String(), Serdes.String()))

Testing scheduled sources

The TopologyTestDriver drives scheduled sources deterministically — no clock waiting. Three controls:

  • triggerScheduledSource(name) — fires a source immediately, once, by name. With no Named, the auto name is scheduled-source-0, -1, … in declaration order.
  • advanceWallClockTime(duration) — advances the test clock; interval (WALL_CLOCK_TIME) and cron sources fire for every boundary crossed.
  • advanceWatermark(timestamp) — advances the event-time watermark; STREAM_TIME interval sources fire for each interval crossed. Cron sources do not fire on watermark advance.

Pass initialWallClockTime to fromBuilder to pin the starting instant for cron and wall-clock assertions. In Kotlin it's a named, defaulted parameter. From Java, fromBuilder is @JvmOverloads over (builder, config, initialWallClockTime) — to set the instant you must also pass the StreamsConfig (there is no two-argument (builder, Instant) overload), so construct one explicitly.

import io.stoatflow.testutils.TopologyTestDriver
import java.time.Instant

val builder = StreamsBuilder()
builder
    .scheduled<String, String>(
        interval = Duration.ofSeconds(10),
        type = PunctuationType.WALL_CLOCK_TIME,
        emitter = { context -> context.forward("key1", "tick") },
    )
    .to("output", Produced.with(Serdes.String(), Serdes.String()))

val driver = TopologyTestDriver.fromBuilder(builder, initialWallClockTime = Instant.ofEpochMilli(0L))
try {
    val output = driver.createOutputTopic("output", Serdes.String(), Serdes.String())

    // Option A: fire once, immediately, by name
    driver.triggerScheduledSource("scheduled-source-0")
    output.readRecord()?.key shouldBe "key1"

    // Option B: advance the clock and let the schedule fire
    driver.advanceWallClockTime(Duration.ofSeconds(10))
    output.readRecord()?.value shouldBe "tick"
} finally {
    driver.close()
}

Execution semantics

A few behaviours worth knowing when you write an emitter:

  • One run at a time. Each scheduled source uses locked-run semantics: if a previous invocation is still running when the next fire is due, the new fire is skipped rather than overlapped. Keep emitters reasonably quick, or expect skipped ticks under load.
  • Records flow through the engine like any source. Emitted records participate in the same exactly-once commit boundaries as Kafka-sourced records — see Exactly-once semantics and the architecture overview.
  • State access is read-only. Use a scheduled source to observe state and emit derived events; mutate state from the processors that handle those events, not from the emitter.
  • STREAM_TIME is data-driven. A STREAM_TIME source only fires as the watermark advances, so it won't fire at all when no records are flowing. Choose WALL_CLOCK_TIME or a cron schedule when you need firing independent of throughput.

See also

The Processor API

Write custom record-by-record logic with Processor and FixedKeyProcessor — forward, access state, schedule punctuators, and register key-based timers.

Serdes and Avro

Configure default and per-operator serdes, use the built-in Serdes, write custom serdes, serialize windowed keys, and wire up Avro with Confluent Schema Registry.