Stage Triggers

A stream trigger runs a Function each time a batch of streamed records closes. It is the only trigger a @stream_publisher has: that decorator takes no trigger_by at all.

from tabsdatak.api import stream_publisher, TableFramesSpec, TableFrameSpec
from tabsdatak.conn.kafka import KafkaStreamSrc


@stream_publisher(
    source=KafkaStreamSrc(
        topic="users",
        data_format="avro",
        schema=AVRO_SCHEMA,
        src_cfg={"kafka.messages_rollover": 500},
    ),
    flush_interval_mins=1,
    output_tables=["users"],
)
def ingest(users: TableFramesSpec) -> TableFrameSpec:
    return concat(users)

How a batch becomes a run

A stream worker consumes the source continuously, buffering records as they arrive. When a rollover threshold trips, the worker writes the buffer out as parquet, stages it, and commits the source's offset. The Function then runs against that staged batch and commits a new Version of its output Tables.

The worker is long-lived and independent of the Function. It keeps consuming while the previous batch is being processed, which is why a slow Function does not drop records.

What closes a batch

Two things, whichever happens first:

• The source's own threshold. Kafka takes kafka.time_rollover_seconds, kafka.size_rollover_mb, and kafka.messages_rollover in src_cfg. The CDC sources take buffer and trigger thresholds in theirs.
• flush_interval_mins on the decorator, in whole minutes, minimum 1.

flush_interval_mins is a ceiling rather than a period. Kafka rolls over at min(time_rollover_secs, flush_interval_mins * 60), so a connector threshold shorter than the flush interval wins, and usually does. Set the flush interval to bound how long a partly-filled batch can sit unprocessed, and set the connector thresholds to control normal batch size.

Available stream sources

Source	System	Output Tables
KafkaStreamSrc	Apache Kafka	one; the topic name is the Table name
MySQLCdcStreamSrc	MySQL	one per tracked Table
PostgresCdcStreamSrc	PostgreSQL	one per tracked Table
Db2CdcStreamSrc	IBM Db2	one per tracked Table

All four are registered as stream sources, which tdk connection types shows in the Mode column as Stream Source. Like ordinary sources they belong to a sources Collection, and the address and credentials live on that Collection's Connection rather than on the source object.

The CDC sources track a list of qualified Table names and resume from a stored position. Postgres additionally requires a logical replication slot, since the WAL it reads is retained on the server until the consumer confirms it. See Postgres CDC, MySQL CDC, and Db2 CDC.

Constraints on the Function

A @stream_publisher differs from the other decorators in three ways worth knowing before you write one:

• It takes one TableFramesSpec parameter per source slot, always in array mode, because a batch is a list of frames rather than a single frame.
• It returns one TableFrame per output Table, the same as a Publisher.
• It must not declare a ctx parameter. It runs inside the runtime's stream loop rather than per-transaction, so there is no TrxCtx at the call boundary and declaring one fails registration.