Scheduled sources: a topology source that isn't a topic

Every source in a Kafka Streams topology is a topic. It is a clean rule, and it has a cost: anything that isn't already in Kafka — the calendar, a snapshot across the keys you hold, a sweep over your own state — needs a job outside the topology to write a trigger into a topic first, or a punctuator bent into a source. The job is its own deployment with its own failure modes and no transactional link to the processing it triggers; the punctuator was never designed for the role, and a great many teams used it that way anyway, because it was the only in-topology option. StoatFlow's scheduled sources make it first-class: StreamsBuilder.scheduled() runs your code on a schedule — read the clock, read your state — and whatever it forwards becomes an ordinary KStream: keyed, processed in parallel on the lanes, and committed in the same transaction as everything else. The same code can reach outside the engine too, to a database or an API; that read sits outside the transaction, and this post stays with the cases the engine can stand behind.

TL;DR

• What: StreamsBuilder.scheduled() — a source node that runs a ScheduledEmitter on an interval or a cron schedule and returns a KStream<K, V>. The emitter is your code, with the clock and a read-only view of every state store.
• Why: without it, time-driven and state-driven work needed an external job and a topic, or a punctuator used as a workaround — serial, on the processing path, outside your commit.
• How: the emitter runs on its own virtual thread; its records enter the engine like any source record, are routed by key to the lanes, and commit with the Kafka-sourced ones in one transaction.
• The catch: the emitter's view of state is read-only, by design; the scan is one thread's work per tick; a run does not yield to live traffic on its own; and a read from outside the engine is not part of the transaction.
• Docs: Scheduled sources.

In this post

• A source that isn't a topic
• What this opens up
• How it works
• A first example: a snapshot across keys
• One use case in depth: the scan-and-act workaround
• Tradeoffs and limits
• Timer, punctuator, or scheduled source

A source that isn't a topic

scheduled() adds a node to the topology with no topic behind it. Two things define it: a schedule and an emitter. The schedule is an interval — on wall-clock time, or on stream time so it follows the watermark — or a cron expression (CronExpression.unix, .quartz or .spring, matching the dialect of the string). The emitter is a ScheduledEmitter<K, V>: a function that receives a ScheduledEmitterContext and calls forward(key, value) zero, one or many times.

What the emitter does in between is up to you. It has two inputs the engine vouches for. The clock — currentWallClockTime() and currentWatermarkMs(). And your own state, through getStateStore(name): a read-only view of any store in the topology, whatever its family. It is ordinary code on a virtual thread, so it can also reach outside — a database, an HTTP endpoint — and a blocking call there parks cheaply and holds up nothing else; but that read is not part of the Kafka transaction its records will commit in, which Tradeoffs and limits spells out. It has one output: keyed records.

The output is the point. What comes back from scheduled() is a KStream<K, V> with nothing special about it. Join it against a KTable. Aggregate or window it. toTable() it and join the live stream against that. merge it with a stream read from Kafka. Send it to a topic. Every operator in the DSL applies, and because the records carry a key, the engine routes them to lanes the way it routes everything else — same key, same lane, in order.

What it replaces: the punctuator hung off an empty topic to get periodic work into the stream, and the cron-plus-producer job that existed only to put a calendar tick or a snapshot request into a topic. What it does not replace: a connector. High-volume ingestion from outside Kafka stays a connector's job, and the caveat above is why.

What this opens up

A short list, deliberately without detail — each of these is a few lines of emitter plus whatever you build downstream, and every one of them reads state the engine already holds.

Events derived from your own state

• Expiry sweeps, reconciliation passes, re-checks against thresholds that have changed since the record arrived — the case the scan-and-act workaround was invented for, covered in depth below.
• Manual TTL and eviction: entries past their timestamp, removed by a paced sweep. A timer per key is the efficient answer when the deadline is known at write time; the sweep is for when it is not, or when the eviction must stay out of the live path — and backpressure and ordering says what "out of the live path" takes. Native store TTL is planned.
• Re-scoring: every open position against today's parameters, every open alert against the current rule set. The deadline was not knowable at ingest, so no timer could have been registered for it.

A decision that needs a view across keys

• Totals and snapshots: open exposure across all accounts, how many are overdrawn right now, the top ten by balance — one pass, a few records, and something no per-key timer can see (the first example below).
• Rankings and leaderboards, with each rank change flowing downstream under its own key.

The calendar as a source

• End-of-day settlement, monthly billing runs, quota resets at the top of the hour, business-day cut-offs — one cron tick that fans out into one record per key.

Records that no topic carries

• Heartbeats and health signals that downstream systems watch for.
• Synthetic load for a demo or a soak test, generated in-topology (the map-filter-runtime example in the repository does exactly this).

And outside data. The same emitter can poll an endpoint or tail a table, one record per row, keyed. It works, and it is the one item here the engine does not yet vouch for: the read happens outside the Kafka transaction, so it is at-least-once from that source however exactly-once the topology is after it. We will write about that pattern when it has the guarantees it deserves; until then, treat it as possible rather than recommended.

What every item shares: the records are first-class, keyed, processed in parallel, and inside the commit. The external job never was; the punctuator workaround was none of the last three.

How it works

At the level that changes what you would build, the path is short.

1. The schedule fires. Interval sources on wall-clock time fire on the clock; on stream time they fire as the watermark advances. Cron sources fire on the clock.
2. The emitter runs on its own virtual thread, off the processing path. Its reads through the read-only view — and anything else it calls — block only that thread. Writes through the view throw, for every store family.
3. Its records enter the engine like source records. The dispatcher routes them by key to the lanes, exactly as it routes records read from Kafka. Same key, same lane — so a scheduled record for acc-1 and a deposit for acc-1 that arrives in the same epoch are processed on one lane, in a defined order.
4. The lanes do the work. Whatever you built downstream of the source runs per record with full read/write state and forwards to sinks.
5. They commit with everything else. The records participate in the same commit barriers as Kafka-sourced records. One barrier closes the epoch; one Kafka transaction commits state, output and offsets for both kinds of record together. Exactly-once holds for a scheduled emission exactly as it holds for a record you consumed — see Exactly-once.

One run at a time, per source: if a tick is due while the previous run is still going, the new tick is skipped, not queued. An exception thrown by the emitter goes through the same processing-exception handler as a processor failure: continue drops that tick's effect, fail aborts the epoch and stops the application.

How the engine hands records from the emitter to the lanes is deliberately kept in the source. The contract is what you rely on: your code in, keyed records out, committed with the stream. Two properties of that hand-off decide what a large run does to the rest of the stream, and both are worth knowing before you write a sweep.

Backpressure and ordering

The emitter cannot outrun the engine. forward() hands records into a bounded buffer — sized by lanes.queue-capacity, 300 by default — and blocks when it is full, so a slow engine slows the emitter and never the reverse; a record then moves onto a lane only when that lane has room. The buffer's depth is the stoatflow.buffer.scheduled.source.size gauge. What this bounds is memory: a run cannot pile up unprocessed records.

A run does not wait behind live traffic. The dispatcher merges scheduled records with Kafka-sourced ones in timestamp order, and a scheduled record is placed at the watermark the run started with. That is the right position for a STREAM_TIME source, and it has a consequence for every other kind: the live records already buffered carry later timestamps, so the run's records go first. Live traffic wins only on equal timestamps. A sweep that emits a million records therefore leads the live stream, in lockstep with the dispatcher, until it is done.

So for low-priority work — a TTL sweep, a reconciliation pass — pace the run yourself: a bounded batch per tick, and a position in state so the next tick continues where this one stopped — range() from the stored key onwards, the position written back downstream with an atomic compute — and the schedule sets the rate. A per-epoch pacing option on the source — at most so many of its records in any one commit — and a background dispatch tier that drains scheduled records only when the live sources are quiet are on the roadmap; until they ship, batch-per-tick is the throttle.

A first example: a snapshot across keys

A risk desk wants two numbers every minute: the open exposure — the sum of every positive balance — and how many accounts are overdrawn. Both are properties of the whole store. No timer can produce them, because a timer fires for one key; no record-driven operator can either, because they have to be true at one instant across keys that update independently. A scheduled source is one read-only pass at that instant.

First, the store and the processor that keeps balances current. Transactions arrive on a transactions topic keyed by account; the processor runs on the account's lane and is the only writer:

private const val ACCOUNTS = "accounts"

/** Applies each transaction to the account's stored balance. Runs on the account's lane. */
class ApplyTransaction : ContextualFixedKeyProcessor<String, Long, Long>() {
    private lateinit var accounts: KeyValueStore<String, Long>

    override fun init(context: FixedKeyProcessorContext<String, Long>) {
        super.init(context)
        accounts = context.getStateStore(ACCOUNTS)
    }

    override fun process(record: FixedKeyRecord<String, Long>) {
        accounts.put(record.key, (accounts.get(record.key) ?: 0L) + record.value)
    }
}

/** The store and the processor that keeps balances current — shared by both examples. */
fun accountsTopology(): StreamsBuilder {
    val builder = StreamsBuilder()

    builder.addStateStore(
        Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore(ACCOUNTS), Serdes.String(), Serdes.Long()),
    )

    builder
        .stream("transactions", Consumed.with(Serdes.String(), Serdes.Long()))
        .processValues({ ApplyTransaction() }, Named.`as`("apply-transaction"), ACCOUNTS)

    return builder
}

Then the source: every minute, one pass, two records, each keyed by what it is:

fun snapshotTopology(): StreamsBuilder {
    val builder = accountsTopology()

    builder
        .scheduled<String, Long>(
            named = Named.`as`("risk-snapshot"),
            interval = Duration.ofMinutes(1),
            type = PunctuationType.WALL_CLOCK_TIME,
            emitter = { context ->
                val accounts = context.getStateStore<ReadOnlyKeyValueStore<String, Long>>(ACCOUNTS)
                var exposure = 0L
                var overdrawn = 0L
                accounts.all().use { entries ->
                    entries.forEach { entry ->
                        if (entry.value > 0) exposure += entry.value else if (entry.value < 0) overdrawn++
                    }
                }
                context.forward("exposure", exposure)
                context.forward("overdrawn", overdrawn)
            },
        ).to("risk-snapshots", Produced.with(Serdes.String(), Serdes.Long()))

    return builder
}

A key per metric is what a snapshot series is, and risk-snapshots is a topic that downstream systems read like any other. Nothing runs on the lanes for this one — the pass is the whole job, and the emitter is the right thread for it: a read-only scan, no writes, off the processing path, committed with the epoch it lands in. In a test, triggerScheduledSource("risk-snapshot") fires it once:

val driver = TopologyTestDriver.fromBuilder(snapshotTopology())
val transactions = driver.createInputTopic("transactions", Serdes.String(), Serdes.Long())
val snapshots = driver.createOutputTopic("risk-snapshots", Serdes.String(), Serdes.Long())

transactions.pipeInput("acc-1", 100_000L)
transactions.pipeInput("acc-2", 25_000L)
transactions.pipeInput("acc-3", -4_000L) // overdrawn

driver.triggerScheduledSource("risk-snapshot")

snapshots.readKeyValuesToList().map { "${it.key}=${it.value}" } shouldBe listOf("exposure=125000", "overdrawn=1")

One use case in depth: the scan-and-act workaround

Punctuators are the right tool for periodic work on the processing path — a heartbeat, a flush of buffered output, a metrics sample, small periodic state maintenance — and StoatFlow runs them with the Kafka Streams API, on a dedicated punctuation lane per sub-topology, with full read/write access to state. What they were never designed for, and were used for anyway, is being a source: with no other way to get non-topic work into the stream, context.schedule(...) became the place for the periodic scan — every N seconds, iterate a state store and act on every entry that is due. It works, and it was always a workaround. The workaround has a shape, and the shape has a cost:

context.schedule(Duration.ofMinutes(1), PunctuationType.WALL_CLOCK_TIME) { now ->
    store.all().use { entries ->
        entries.forEach { entry ->
            if (isDue(entry.value, now)) {
                val result = settle(entry.value)    // the business logic
                store.put(entry.key, result)         // the state mutation
                context.forward(entry.key, result)   // the output
            }
        }
    }
}

Two jobs are fused in that loop: deciding what needs doing — the scan, which has to see every entry — and doing it, which needs one entry at a time. In Kafka Streams the loop runs on the stream thread, in the same loop that calls process(): while it iterates, no record on any task that thread owns moves, and it runs once per task, so a million entries across twelve partitions are scanned twelve times, serially. In StoatFlow the record lanes keep processing while a punctuator runs — but the punctuation lane takes part in the commit barrier, and in the default BLOCKING mode the next commit waits for it. Under exactly-once, output is visible only when the transaction commits, so a five-second scan adds five seconds to the end-to-end latency of every record on every lane in that epoch — and the six settlements inside it still ran one after another on one lane. None of this is a fault in punctuators. It is what happens when a callback built for small periodic work is asked to be an ingest path.

A scheduled source splits the two jobs. Deciding happens once, on one thread, against the whole store. Acting happens per key, on that key's lane, with full read/write state — in parallel across lanes, and ordered against every other record for that key. That is also why the emitter's view is read-only by design: a write from the emitter's thread would be unordered against the records for that key arriving on its lane, so the write belongs downstream, on the lane. The emitter observes; the lane acts.

Left: six settlements in series on the punctuation lane, commit N waiting for all six, and the gold band is everything the four lanes processed meanwhile — done, but invisible downstream until the loop ends. Right: the same six on three lanes, by key, and commit N follows the busiest lane after three. The lane numbers come from the engine's real hash; only the layout was designed.

The example: the same accounts, and now they accrue interest daily. At midnight every account with a balance is settled — interest computed and written, a statement emitted. This is not a timer case. A timer fires for one key at one instant you knew when the record arrived; here the trigger is a calendar event that fans out to every account at once, and which accounts are involved is a question only the store can answer. The store and ApplyTransaction are the ones from the first example.

The scheduled source fires at midnight, scans the store through the read-only view, and emits one record per account with something to settle. It decides; it does not act:

val builder = accountsTopology()

val settlements =
    builder.scheduled<String, Long>(
        named = Named.`as`("eod-scanner"),
        cron = CronExpression.unix("0 0 * * *"), // every day at midnight
        emitter = { context ->
            val day = context.currentWallClockTime() / 86_400_000L
            val accounts = context.getStateStore<ReadOnlyKeyValueStore<String, Long>>(ACCOUNTS)
            accounts.all().use { entries ->
                entries.forEach { entry ->
                    if (entry.value != 0L) context.forward(entry.key, day)
                }
            }
        },
    )

The processor that settles one account is attached downstream of the source, so it runs on the account's lane with read/write access to the same store — serialised against that account's transactions, in parallel with every other account's settlement:

/** Accrues one day's interest for one account. Runs on that account's lane. */
class SettleAccount(private val dailyRateBps: Long) : ContextualProcessor<String, Long, String, String>() {
    private lateinit var accounts: KeyValueStore<String, Long>

    override fun init(context: ProcessorContext<String, String>) {
        super.init(context)
        accounts = context.getStateStore(ACCOUNTS)
    }

    override fun process(record: Record<String, Long>) {
        val balance = accounts.get(record.key) ?: return
        val interest = balance * dailyRateBps / 10_000
        accounts.put(record.key, balance + interest)
        context().forward(record.key, "day=${record.value} balance=${balance + interest} interest=$interest")
    }
}

settlements
    .process({ SettleAccount(dailyRateBps = 10) }, Named.`as`("settle-account"), ACCOUNTS)
    .to("daily-statements", Produced.with(Serdes.String(), Serdes.String()))

A deposit for acc-1 that lands at one second past midnight is routed to the same lane as acc-1's settlement record. Whichever the dispatcher handed on first runs first, and the settlement reads the balance the lane has at that moment — no lock, no read-modify-write race, because the ordering is the lane's. The test has the same shape as the snapshot's: pipe a few transactions, triggerScheduledSource("eod-scanner"), read the statement, check the balance moved. (advanceWallClockTime fires cron boundaries too, but cron resolves in the system default zone, so a test that crosses midnight is a test of the machine's zone.)

Both examples in this post run green against the current build; the code is lifted from those tests unchanged.

Porting the workaround from Kafka Streams is mechanical: the loop body splits where it stops reading and starts writing — everything before is the emitter, everything after is the downstream processor, and the key you would have written to the store is the key you forward. The punctuator's interval becomes the source's interval or cron. Punctuators doing what they are for — a heartbeat, a flush, a metrics sample — stay exactly as they are.

Tradeoffs and limits

• The emitter cannot write. Every store family — key-value, window, session, timestamped, versioned, headers-aware — is wrapped read-only, and put or delete through the view throws UnsupportedOperationException. This is the point of the design, but it means a "scan and mark as done" loop becomes two pieces, not one. The mark belongs in the downstream processor.
• A read from outside the engine is not in the transaction. The emitter can query a database or an API; that call happens outside the Kafka transaction its records will commit in, so exactly-once holds for the records' effects and not for the read. It is at-least-once from that source until the engine can do better — which is why this post's examples read state, not the network.
• One pass per tick, not a firehose. One virtual thread, one run per tick, through a bounded buffer. High-volume ingestion from outside Kafka stays a connector's job.
• Bound the run. A slow emitter — a call that hangs, a scan that grows — holds the locked run, and every tick due meanwhile is skipped until it returns. Put timeouts on anything it calls and a budget on what it scans; the engine will not do it for you.
• A run does not yield to live traffic on its own. Its records are placed at the watermark the run started with, ahead of the live records buffered at that moment; the buffer bounds memory, not latency. Pace low-priority sweeps with a bounded batch per tick and a position in state, and watch stoatflow.buffer.scheduled.source.size.
• The scan is still one thread's work per tick. Parallelism applies to the act, not the decide. An all() over ten million entries every minute is a real cost on one virtual thread. Bound it: range() and prefixScan() if the keys carry what you need, or have the processors maintain a small index store keyed by due date that the scanner reads instead of the main store. Keep deciding cheap; the lanes will absorb the doing.
• Overlapping ticks are skipped, not queued. If a run is still going when the next tick is due, the tick is dropped and logged. A run slower than its own interval silently halves its own frequency — watch for it.
• Failures route like a processor's. An exception from the emitter goes to the processing-exception handler. continue drops that tick — the next one re-derives everything from state; fail aborts the epoch and stops the application.
• STREAM_TIME is data-driven. It fires as the watermark advances. With no records flowing it does not fire at all; use wall-clock or cron for anything that must happen regardless of throughput.
• An emission has no Kafka offset. If the process dies before the epoch commits, the emitted records are gone with the transaction — and so are the state writes the lanes made from them, since they were in the same epoch. The next tick re-derives its emissions from state, which has rolled back with everything else. That is the right behaviour, and it only holds if the emitter is a function of state: "every account with a balance", "every entry past its expiry" — not "the ones I have not emitted yet".
• A tick that falls while the process is down is not made up. On start, a cron source computes its next fire time from now; a midnight the application slept through does not fire on restart. If the rule requires it, keep a "last settled day" in state and let the next tick settle the gap. An interval source simply resumes.
• Cron resolves in the system default zone. Pin the container's zone, or you have pinned it to wherever the image was built.
• No Kafka Streams equivalent. scheduled() is a StoatFlow extension — see the compatibility matrix. Code that uses it does not port back.

Timer, punctuator, or scheduled source

All three do time-driven work; they answer different questions.

Each row is the right tool for its question. If the record that arrives already tells you when to act on it, register a timer for that key and never iterate the store to find it. If the work is small and belongs on the processing path, a punctuator is exactly right. The scheduled source's job is the third question — and, by answering it, to stop the other two being bent into a source.

The full API — interval and cron overloads, the emitter context, naming, and the test driver controls — is on Scheduled sources.