SQL Topics & Consumer Groups

Use SQL to create topics, attach source tables, inspect offsets, and do simple topic reads inside KalamDB.

For production workers, the HTTP or SDK consume APIs are still a better fit when you need explicit ack timing and partition selection. The SQL surface is best for setup, inspection, debugging, and simple single-partition flows. SQL consumer groups resume from persisted offsets too, and grouped SQL consumes require an explicit ACK after processing.

Why use groups

Use GROUP when a worker should keep its place in a topic.

Without GROUP, a consume is stateless. Every call starts from the FROM position you requested, so it is good for inspection and replay but not for a durable worker cursor.

With GROUP, KalamDB stores progress in system.topic_offsets for (topic_id, group_id, partition_id). That gives you:

• resume-after-restart behavior for the same logical worker or worker fleet
• independent read positions for different consumers on the same topic
• a way to replay with one group while production workers keep their own offsets

There is no separate CREATE GROUP command. A group is created the first time you consume or acknowledge offsets with a new group_id.

Stateless reads vs durable reads

Use no group when you want a pure viewer:

1CONSUME FROM demo.message_events FROM EARLIEST LIMIT 100;

That query does not store progress. Run it again and you can replay the same range again.

Use a group when you want a named cursor:

1CONSUME FROM demo.message_events2GROUP 'ui-debug-admin'3FROM LATEST4LIMIT 100;

That query creates or resumes the ui-debug-admin cursor for partition 0.

Recommended starting point

Use a single-partition topic until multi-partition behavior is fully finished:

1CREATE TOPIC app.user_events PARTITIONS 1;

CREATE TOPIC requirements today:

• topic names must be namespace-qualified, like app.user_events
• the namespace must already exist
• CREATE TOPIC IF NOT EXISTS is not currently supported
• topic creation requires dba or system role

CREATE TOPIC

1CREATE TOPIC <topic_name>2[PARTITIONS <count>]3[WITH (retention_seconds = <seconds|NULL>, retention_max_bytes = <bytes|NULL>)];

When a topic is created, KalamDB persists it in system.topics and adds it to the in-memory route cache immediately.

Retention options on CREATE TOPIC are optional:

• omit an option to use the server [topics] default
• set an option to NULL to disable that limit for this topic
• retention_max_bytes is enforced per partition, not across the whole topic

Examples:

1CREATE TOPIC app.user_events PARTITIONS 1;2 3CREATE TOPIC app.user_events4PARTITIONS 15WITH (retention_seconds = 86400, retention_max_bytes = 268435456);6 7CREATE TOPIC app.audit_events8PARTITIONS 19WITH (retention_seconds = 3600, retention_max_bytes = NULL);

In the last example, messages are deleted once they are older than one hour, and there is no byte cap for the topic partitions.

Topic retention

Topic retention is a topic-level policy stored in system.topics. It is independent of consumer group ack state.

Current behavior:

• retention_seconds deletes messages older than the configured age
• retention_max_bytes trims the oldest retained messages until the partition is back under the byte cap
• if both limits are active, age-based pruning runs first and byte-based pruning runs after that
• if both limits are NULL, retention is disabled for the topic
• retained message deletion does not rewrite offsets or move consumer group cursors

The default server configuration is 7 days and 1 GiB per partition unless you override it in the [topics] config block:

1[topics]2default_retention_seconds = 6048003default_retention_max_bytes = 10737418244retention_check_interval_seconds = 36005retention_batch_size = 10000

ALTER TOPIC retention

Use ALTER TOPIC when you want to change retention after the topic already exists.

1ALTER TOPIC <topic_name>2SET RETENTION WITH (retention_seconds = <seconds|NULL>, retention_max_bytes = <bytes|NULL>);3 4ALTER TOPIC <topic_name> CLEAR RETENTION;

SET RETENTION only changes the options you specify. Omitted options keep their current value.

Examples:

1ALTER TOPIC app.user_events2SET RETENTION WITH (retention_seconds = 604800, retention_max_bytes = 1073741824);3 4ALTER TOPIC app.user_events5SET RETENTION WITH (retention_seconds = NULL);6 7ALTER TOPIC app.user_events8SET RETENTION WITH (retention_max_bytes = 536870912);9 10ALTER TOPIC app.user_events CLEAR RETENTION;

Use CLEAR RETENTION when you want to disable both limits in one step.

How retention affects reads

Retention removes stored messages, but it does not renumber topic offsets.

• CONSUME FROM ... FROM EARLIEST starts at the earliest currently retained offset
• explicit offsets below that low watermark fail because those messages are no longer available
• a consumer group with a committed offset below the earliest retained offset must be reset before it can continue

Example recovery flow for a lagged group:

1RESET CONSUMER GROUP 'backfill-worker' ON app.user_events TO 250;2 3CONSUME FROM app.user_events4GROUP 'backfill-worker'5FROM EARLIEST6LIMIT 100;

That reset changes the next offset for the group. The FROM clause in later grouped reads does not override an already-committed group cursor.

ADD SOURCE

1ALTER TOPIC <topic_name>2ADD SOURCE <table_name_or_namespace.table_name>3ON <INSERT|UPDATE|DELETE>4[WHERE <filter_expression>]5[WITH (payload = '<key|full|diff>')];

The parser accepts key, full, and diff, but only full is supported today. key and diff are still in development.

Current route behavior:

• a route matches by source table and operation
• an optional row-local WHERE predicate is evaluated before a message is published
• one changed row becomes one topic message
• route registration updates system.topics and the in-memory route cache immediately
• ALTER TOPIC ... ADD SOURCE requires dba or system role

You can add separate routes for INSERT, UPDATE, and DELETE:

1ALTER TOPIC app.user_events ADD SOURCE app.users ON INSERT WITH (payload = 'full');2ALTER TOPIC app.user_events ADD SOURCE app.users ON UPDATE WITH (payload = 'full');3ALTER TOPIC app.user_events ADD SOURCE app.users ON DELETE WITH (payload = 'full');

Payload modes

Only payload = 'full' has behavior you should rely on today. key and diff are still in development:

1ALTER TOPIC app.user_events2ADD SOURCE app.users3ON INSERT4WITH (payload = 'full');

What those modes mean in the current code:

If you want predictable payloads today, use payload = 'full'. Treat key and diff as in development.

WHERE filters

Use WHERE when only some row changes should publish to the topic. The predicate is evaluated against the row routed for the selected operation.

That is useful for worker queues that should only wake up on a subset of writes. For example, route only cancelled tasks into a cleanup worker topic:

1ALTER TOPIC app.task_cancellations2ADD SOURCE app.tasks3ON INSERT4WHERE cancelled = true5WITH (payload = 'full');6 7ALTER TOPIC app.task_cancellations8ADD SOURCE app.tasks9ON UPDATE10WHERE cancelled = true11WITH (payload = 'full');

With those two routes:

• a task inserted with cancelled = true publishes immediately
• a task that starts active and is later updated to cancelled = true publishes on that update
• rows where cancelled = false are ignored by these routes

Keep the predicate row-local. Subqueries are rejected for route filters.

CONSUME FROM

1CONSUME FROM <topic_name>2[GROUP '<group_id>']3[FROM <LATEST|EARLIEST|offset>]4[LIMIT <count>];

Examples:

1CONSUME FROM app.user_events FROM EARLIEST LIMIT 10;2 3CONSUME FROM app.user_events4GROUP 'ai-worker'5FROM EARLIEST6LIMIT 10;

Current SQL consume behavior:

• SQL CONSUME reads partition 0 only
• the result columns are topic_id, partition_id, offset, key, payload, timestamp_ms, user_id, and op
• payload is a binary column containing the stored message bytes
• user_id is the canonical producing user id when the message came from a USER-table scoped write
• without GROUP, the read is stateless
• with GROUP, KalamDB first checks system.topic_offsets and resumes from the committed offset when one exists
• if a group has no committed offset yet, the FROM clause decides the initial position
• with GROUP, the returned range is claimed in memory until it is acknowledged or the visibility timeout expires
• use ACK to persist progress after processing a grouped batch

If you need explicit ack timing or partition selection, use the HTTP Topic API.

New group vs existing group

The important rule is:

• for a brand-new group, FROM decides where the first read starts
• for a group that already has a committed offset, KalamDB resumes from that offset and ignores FROM

Examples:

New group that tails only future messages:

1CONSUME FROM demo.message_events2GROUP 'ui-debug-admin'3FROM LATEST4LIMIT 100;

Use that when you create a fresh debug group and only want messages written after the group starts.

New group that replays the backlog:

1CONSUME FROM demo.message_events2GROUP 'backfill-worker'3FROM EARLIEST4LIMIT 100;

Use that when a new worker should process existing messages from the beginning.

Existing group that already consumed part of the topic:

1CONSUME FROM demo.message_events2GROUP 'backfill-worker'3FROM LATEST4LIMIT 100;

Even though this query says FROM LATEST, KalamDB resumes backfill-worker from its committed offset. The group already has a cursor, so the FROM clause no longer changes the starting point.

If you created a new group on the fly and expected to see existing messages, do not use FROM LATEST. Use FROM EARLIEST or FROM <offset> for that first consume.

How to use groups safely

Use the same group id for one logical worker fleet:

1CONSUME FROM app.user_events GROUP 'email-worker' FROM EARLIEST LIMIT 50;

Use a different group when you want an independent cursor:

1CONSUME FROM app.user_events GROUP 'audit-replay' FROM EARLIEST LIMIT 50;

Use no group when you want a viewer that never advances offsets:

1CONSUME FROM app.user_events FROM EARLIEST LIMIT 50;

Use a throwaway group for one-off debugging when you want group semantics without affecting a real worker group:

1CONSUME FROM app.user_events2GROUP 'debug-2026-04-24-admin'3FROM EARLIEST4LIMIT 50;

That lets you inspect with a separate durable cursor while production groups keep their own positions.

ACK

1ACK <topic_name>2GROUP '<group_id>'3[PARTITION <partition_id>]4UPTO OFFSET <offset>;

Current ack behavior:

• ACK writes to system.topic_offsets
• acknowledgements are monotonic, so a lower offset does not move a group backward
• ACK is per topic, group, and partition
• ACK requires service, dba, or system role

If you are using SQL CONSUME ... GROUP, issue ACK after you finish processing the returned rows. If the caller does not ACK before the topic visibility timeout, the range can be delivered again.

RESET CONSUMER GROUP

1RESET CONSUMER GROUP '<group_id>'2ON <topic_name>3[PARTITION <partition_id>]4TO <next_offset>;

Use reset when you deliberately want to move a durable group cursor. This is an admin-only operation and affects only the named topic, group, and partition.

Replay a debug group from the beginning:

1RESET CONSUMER GROUP 'ui-debug-admin1' ON blog.summarizer TO 0;2 3CONSUME FROM blog.summarizer4GROUP 'ui-debug-admin1'5FROM EARLIEST6LIMIT 100;

Move a group so the next read starts at offset 250:

1RESET CONSUMER GROUP 'backfill-worker' ON app.user_events PARTITION 0 TO 250;

For inspection without moving or resetting any group, omit GROUP:

1CONSUME FROM blog.summarizer FROM 0 LIMIT 100;

CLEAR TOPIC

1CLEAR TOPIC <topic_name>;

CLEAR TOPIC keeps the topic definition but schedules a background cleanup job to delete stored messages and consumer offsets.

DROP TOPIC

1DROP TOPIC <topic_name>;

DROP TOPIC removes the topic definition and route cache entry immediately, then schedules the same background cleanup job for stored messages and offsets.

Inspect topic state

Topic metadata and consumer progress are queryable:

1SELECT topic_id, partitions, routes, created_at, updated_at2FROM system.topics3ORDER BY topic_id;4 5SELECT topic_id, group_id, partition_id, last_acked_offset, updated_at6FROM system.topic_offsets7ORDER BY topic_id, group_id, partition_id;

The message log itself is internal. You inspect messages with CONSUME, not by querying a system.topic_messages table.

Practical guidance

• use PARTITIONS 1
• use payload = 'full'; it is the only payload mode with fully documented behavior today
• use a separate source table and result table for agent flows so you do not create loops
• treat key and diff as still in development, and treat route WHERE and custom partitioning as incomplete for now