Trigger off a Kafka Stream
A Kafka stream source consumes one topic continuously and emits a single Tabsdata table named after that topic. What runs the publisher is neither a clock nor an upstream commit but a rollover: the stream worker buffers messages, and when a rollover threshold trips it writes the buffer out as parquet and hands the batch to the publisher to ingest.
A stream publisher therefore takes no trigger at all.
The brokers live on the collection's connection, not in the function, so the collection
has to exist with a kafka-topic-in connection bound to it before a stream publisher can
be registered there. Only the bootstrap servers are required. A broker that takes SASL
needs credentials alongside them, and one that does not takes none.
Needs a stream publisher, and a source collection with a Kafka connection to hold it. Four values left, then I generate and register. Reply with them one per line, or point me at a file that has them:
- collection name
- bootstrap servers
- topic
- message format and schema
Read 2 files, called tabsdata 2 times (ctrl+o to expand)
Schema read from ~/clicks.json. No credentials in the file, so the connection goes out unauthenticated: credentials is optional on a Kafka connection, and this broker takes no SASL. Reading back: broker.internal:9092, topic clicks, format json, collection events.
Called tabsdata 3 times (ctrl+o to expand)
Collection events created in group sources, with a KafkaStreamSrcConn to broker.internal:9092. Connection check passed and the topic exists.
Generating — KafkaStreamSrc on the clicks topic. One topic per source, and the topic name becomes the table name, so there is exactly one output table.
Called tabsdata 3 times (ctrl+o to expand)
The stream publisher, registered in events:
@stream_publisher(
source=KafkaStreamSrc(
topic="clicks",
data_format="json",
schema=CLICK_SCHEMA,
),
flush_interval_mins=5,
output_tables=["clicks"],
)
def read_clicks(clicks: TableFramesSpec) -> TableFramesSpec:
return clicks
Called tabsdata 2 times (ctrl+o to expand)
Execution committed, 0 failures.
events/clicks is live — the worker is consuming, and the first batch lands at the next rollover:
┌───────────────┬───────────┐ │ field │ value │ ├───────────────┼───────────┤ │ topic │ clicks │ ├───────────────┼───────────┤ │ format │ json │ ├───────────────┼───────────┤ │ output table │ clicks │ ├───────────────┼───────────┤ │ flush ceiling │ 5 min │ ├───────────────┼───────────┤ │ worker │ consuming │ └───────────────┴───────────┘
schema is required, and in schema-registry mode it is cross-checked against the
registry. The function takes one input slot per source, always in array mode, and
declares no transaction context: it runs inside the runtime's stream loop rather than
per-transaction.
What closes a batch
Rollover fires on whichever threshold trips first, the connector's own or the
decorator's fallback. Kafka takes min(time_rollover_secs, flush_interval_mins * 60),
so the flush interval is a ceiling and the connector's window is usually the one that
fires.
| Key | Closes the batch when |
|---|---|
kafka.time_rollover_seconds | That many seconds have passed |
kafka.size_rollover_mb | The buffer reaches that size |
kafka.messages_rollover | That many messages have accumulated |
The flush interval takes whole minutes, minimum 1.
Called tabsdata 3 times (ctrl+o to expand)
Set kafka.time_rollover_seconds to 30 in the source config. The 5-minute flush interval stays as the ceiling, and the 30-second window now fires first.
source=KafkaStreamSrc(
topic="clicks",
data_format="json",
schema=CLICK_SCHEMA,
src_cfg={"kafka.time_rollover_seconds": 30},
),
Messages that fail to deserialize
Set a dead-letter topic on the source and messages that cannot be deserialized are routed there instead of failing the batch.
What to do next
Trigger off CDC in an SQL Database covers the other three stream sources, which follow the same rollover model against a database log.
See Stage Triggers for how a batch closes and becomes a run, and Kafka for the full connector setup.