Stream and queue consumers

Most sources are polled on an interval. Stream and queue sources are read continuously by long-running consumers instead, so a record lands seconds after it is published. Each message becomes a record (records mode) or a ticket (pipeline mode), or a deletion.

Source types#

Kafka and Kinesis also take Start a new partition at (or shard): earliest (default) or latest.

Kafka needs the confluent-kafka package and Kinesis boto3 (pip install 'ticketiq[streams]'). NATS needs nats-py. Without the package, the source is refused with a message that says so. Kafka broker and NATS server names are vetted before each connect.

Enter secrets as secret:// references (see Secrets).

What a message becomes#

Set the source's sync mode to records or pipeline. Records mode takes the same fields as any records source: Record type, the column that identifies a record, the title column and an optional schema (name:type, …).

• A JSON object becomes a record, upserted by key, or a ticket draft run through the agent pipeline.
• A body that is not a JSON object (or, in pipeline mode, has no title) is dropped and counted as a rejection.
• A Kafka record with a key and no value, or a message {"_deleted": true} or Debezium op: "d", is a deletion. See Deletion propagation.

Stream and queue sources are not synced: Sync refuses them (422) and the interval scheduler skips them. Knowledge-base mode is not offered for them.

How Kafka and Kinesis are consumed#

A stream is an ordered log per partition (Kafka) or shard (Kinesis), and the reader keeps its own place. Liyagent keeps that place itself rather than in a broker consumer group:

• Workers. The leader keeps TICKETIQ_STREAM_WORKERS_PER_SOURCE (default 2) worker jobs queued or running per stream source. Any app process's job worker can claim one, so with two app processes the work is shared, and a worker whose process dies is resumed by another.
• Leases. A worker reads only the partitions it holds a lease on (stream:<source>:<partition>). It takes free partitions up to its fair share and gives back any above it. A partition left free for three passes is taken whatever the share, so a dead process's partitions move within a lease period.
• Checkpoints. Each partition's position is kept in the stream_checkpoints table in Postgres, never the broker. It is read every pass and written after a batch is applied, only up to the last message of the partition applied in order.
• At-least-once, applied idempotently. A crash between applying and checkpointing re-reads the batch. Messages at or below the stored position are dropped; records upsert by key and unchanged rows are not written again; pipeline mode skips an external id it already has; a write and a delete of one key in a batch end as the later one says.

A worker that lost a lease drops that partition's messages unapplied, and the new holder reads them.

TICKETIQ_STREAM_CONSUMERS_ENABLED (default true) turns stream consumers on; with no Kafka or Kinesis source they do nothing.

How queues are consumed#

SQS, Pub/Sub and NATS hand each message to one reader, so the leader runs one consumer job per queue source, started about once a minute while TICKETIQ_QUEUE_CONSUMERS_ENABLED (default true) is on.

• Acknowledgement. A batch is stored first, then acknowledged. If storing fails, the message is left unacknowledged for the queue to deliver again. A body that can never parse is acknowledged and counted as a rejection.
• Pulls. TICKETIQ_QUEUE_BATCH_SIZE (default 10) messages per pull, waiting up to TICKETIQ_QUEUE_WAIT_SECONDS (default 20).
• Failures. A consumer that fails TICKETIQ_QUEUE_MAX_CONSECUTIVE_FAILURES (default 5) pulls in a row stops, and is started again on the next pass a minute later. Stream workers stop on the same count, backing off between passes.

Consumers are ordinary jobs: they show in the jobs list with their progress and can be cancelled. Set consume: off in a source's config to skip it.

See what a consumer is doing#

Open Connect › Ingestion (/ingestion). Each queue or stream source shows its consumer's live state: running or retrying, the last pull, messages received and acknowledged (applied, for streams), the partitions this process holds, and the last error.

Managing sources needs sources.write.

Quality and lineage#

Every 300 seconds, or every 1,000 messages, a consumer writes one quality batch for what it read since the last one, just as a sync writes one per run:

• rows read and rows that failed,
• each field's null rate and type, and the values that did not fit their type,
• drift against the previous batch: fields added, removed or retyped, and null-rate jumps of TICKETIQ_INGEST_NULL_JUMP (default 0.25) or more,
• a status: ok, drift, degraded, empty, failed or cancelled.

Create an alert rule on ingest_rows (a heartbeat: no rows means no data), ingest_null_rate or ingest_drift under Incidents › Rules, or with POST /api/alerts/rules. Rules are scoped to the fleet or a tenant, not to one source.

With ontology lineage on, each quality window is also one lineage line: the source was read, and in pipeline mode each ticket it opened was written. See Ontology.

Limits#

• Kafka offsets are not committed back to the broker, so broker-side lag tools do not see Liyagent's progress.
• Kinesis enhanced fan-out and resharding lineage (draining parent shards before children) are not handled.
• NATS connections are raw TCP, outside the HTTP egress policy. Server names are checked before every connect, but the client resolves them again itself.
• Change data capture straight from a database is not built. A SQL source's watermark is the incremental path.