# Stream and queue consumers

Read Kafka topics, Kinesis streams, SQS, Pub/Sub and NATS continuously into records or tickets, at least once, with quality alerts per window.

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

| Type | Reads | Configuration |
| --- | --- | --- |
| `kafka` | An Apache Kafka topic | **Bootstrap servers**, **Topic**, optional **Security protocol** (`PLAINTEXT`, `SSL`, `SASL_PLAINTEXT`, `SASL_SSL`), **SASL mechanism** (`PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512`), SASL user and password |
| `kinesis` | An Amazon Kinesis stream | **Stream name**, **Region**, an access key id and secret, or blank to use the instance's credential chain |
| `sqs` | An Amazon SQS queue | **Queue URL**, an access key or the credential chain, visibility timeout |
| `pubsub` | A Google Cloud Pub/Sub subscription | **Project id**, **Subscription**, an OAuth connection |
| `nats` | NATS JetStream | **Server URL(s)**, **Stream**, **Subject**, **Durable consumer name**, a token or user and password |

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](/docs/connect/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](/docs/connect/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](/docs/operate/scaling) 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.

| Question | Endpoint | Permission |
| --- | --- | --- |
| Every source with its mode, latest batch and consumer state | `GET /api/ingestion/overview` | `sources.read` |
| Quality batches | `GET /api/ingestion/batches` | `sources.read` |
| Records read | `GET /api/records`, `GET /api/records/types` | `sources.read` |

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](/docs/connect/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.