Kafka clients

--kafka <host:port> makes a node speak the Kafka protocol, on port 9092 by convention. A Kafka client connects to that address as it would to a broker. What it produces lands in a TessariDB topic, and what it fetches comes from one. Nothing sits between the protocol and the store.

tessaridb ./data --serve 0.0.0.0:9080 --kafka 0.0.0.0:9092

A producer and a consumer, unchanged

This is ordinary client code for a broker. Only the address and the credentials are the node's:

from confluent_kafka import Consumer, Producer

settings = {
    "bootstrap.servers": "node.example:9092",
    "security.protocol": "SASL_PLAINTEXT",   # SASL_SSL once the node has a certificate
    "sasl.mechanism": "PLAIN",
    "sasl.username": "billing",              # a TessariDB user
    "sasl.password": password,
}

producer = Producer(settings)
for customer, total in [("c-17", "120.00"), ("c-42", "8.50"), ("c-17", "33.10")]:
    producer.produce("shop.orders.orders", key=customer, value=total, headers={"source": b"web"})
producer.flush(10)

consumer = Consumer({**settings, "group.id": "billing", "auto.offset.reset": "earliest"})
consumer.subscribe(["shop.orders.orders"])
message = consumer.poll(10)
print(message.partition(), message.offset(), message.key(), message.value())
consumer.commit(asynchronous=False)

The topic has to exist first. Producing to an undeclared topic does not create one, and the client reports the topic unknown until it times out. Declare it in TessariQL, or create it with an admin client's CreateTopics:

DEFINE NAMESPACE IF NOT EXISTS shop;
USE NAMESPACE shop;
DEFINE DATABASE IF NOT EXISTS orders;
USE DATABASE orders;
DEFINE TOPIC orders PARTITIONS 3 BY customer;

How names map

A Kafka topic name is <namespace>.<database>.<topic>, so shop.orders.orders is the topic orders in the database shop.orders. Each part is an identifier, so the dots cannot be ambiguous.

--kafka-database shop.orders (or TESSARIDB_KAFKA_DATABASE) lets a bare name like orders be read in that database as well. That is what most applications want, because their configuration already holds bare names.

A name that is neither form is answered INVALID_TOPIC_EXCEPTION, and one naming no topic is answered UNKNOWN_TOPIC_OR_PARTITION.

What a record becomes

A Kafka record is stored as one topic message of this shape:

READ FROM orders PARTITION 1 LIMIT 1;

1: { appended: datetime '2026-10-22T11:45:47.225Z', position: 1, value: { headers: [{ key: 'source', value: 0x776562 }], key: 0x632d3137, timestamp: datetime '2026-10-22T11:45:47.184Z', value: 0x3132302e3030 } }

It works in the other direction too. A message appended by TessariQL, or by a RESP client's XADD, is fetched by a Kafka consumer with no key, its value encoded as JSON, and the instant it was appended as its timestamp:

(0, 0, None, b'{"customer":"c-42","total":19.99}')

Partitions

A topic declared without PARTITIONS is one partition, partition 0. DEFINE TOPIC orders PARTITIONS n BY field splits it, and each partition keeps its own order, as a Kafka partition does.

A Kafka client picks the partition for each record and the node writes it there. A message appended by TessariQL is placed by BY: the field's value is hashed with murmur2, the hash the Java client uses by default, so a key lands in the same partition whichever side wrote it.

Clients built on librdkafka (Python, Go, .NET, C) default to a different hash. To keep their records in step with the store's own appends, set:

producer = Producer({**settings, "partitioner": "murmur2_random"})

With that setting c-42 landed in partition 0 and c-17 in partition 1 on a three-partition topic, the same partitions CREATE orders = { customer: … } chose.

Consumer groups and offsets

Groups work as clients expect: join, sync, heartbeat, leave, rebalance, offset commit and fetch, DescribeGroups, ListGroups, DeleteGroups.

A group's committed offset is stored as a named reader position, under the reader name kafka:<group>, on each partition. TessariQL sees it:

INFO FOR TOPIC orders;

{ by: 'customer', each: [ …, { consumers: { 'kafka:billing': { lag: 0, position: 3 } }, first: 1, last: 3, partition: 1 }, … ], partitions: 3, … }

So READ FROM orders FOR CONSUMER 'kafka:billing' continues where the Kafka group stopped, and lag is the topic's last position minus the group's.

A new group with auto.offset.reset = earliest starts at the first message the topic still holds. Membership and assignments live in the node's memory; after a restart clients re-join as they would with a broker, and their committed offsets are still in the store.

Signing in

The node takes SASL PLAIN and checks the user name and password against the store's own users. Each connection then runs as that user, under that user's grants:

A refused password is answered and the connection is closed, so no second password can be tried on it. The store's sign-in throttling applies as on every other surface.

SCRAM is refused at the handshake. The store keeps Argon2id password hashes, and a SCRAM verifier would be a second copy of every password. A client asked to use SCRAM reports broker's supported mechanisms: PLAIN.

A store with no users is open, as it is everywhere else, and a client may connect without SASL.

TLS

A node given --tls-cert and --tls-key serves the Kafka port over TLS 1.3 with the same certificate as its other surfaces. Set security.protocol = SASL_SSL and give the client the issuing authority, as for any broker. --require-client-tls refuses to start a node whose Kafka port would otherwise be in the clear. See encryption in transit.

Behind a proxy, and on a cluster

A Kafka client reconnects to whatever address the node advertises in Metadata. By default that is the address the client reached.

Behind a port mapping or a load balancer, set --kafka-advertise host:port to the address clients can reach, or they connect once and then fail.

On a cluster, each member says where its Kafka port is, and Metadata lists every member as a broker:

DEFINE REPLICA second AT 'db-2.internal:9180'
    CLIENTS AT 'db-2.example:9080' KAFKA AT 'db-2.example:9092'
    NODE '9f2c4e1a70bb43d5a1c6e2f480937d55' ROLES serving, coordinating;

ALTER REPLICA second KAFKA AT 'db-2.example:9092' (or KAFKA NONE) amends a row in place.

Each partition's leader is the node leading its range. A produce sent to another node is carried to that leader. The group coordinator is the namespace's leader, so one node coordinates each group.

Producers

• Idempotence (enable.idempotence) is honoured. A producer id is issued and duplicates are dropped per partition within Kafka's five-batch window. The node keeps that state in memory, so after a restart the client initialises again.
• Transactions (transactional.id) are refused with TRANSACTIONAL_ID_AUTHORIZATION_FAILED. A topic's messages are committed one produce at a time.
• Compression: gzip, snappy, lz4 and zstd batches are all read.
• acks: acks=0 is answered with nothing. acks=1 and acks=all are answered once the append is durable at the store's acknowledgement level.

The API keys served

ApiVersions reports exactly these, and a client negotiates within them:

A request for any other key or version closes the connection, which is how a broker answers one it does not know.

Next

• Topics — what the store keeps for each message, groups, retention and the lapsed note.
• RESP clients — the other protocol a node speaks.