[jubrad]

Proof of concept, not for merge as is. Opened as a draft to share the design and the code.

A KAFKA LISTENER source turns a cluster replica into a one-broker Kafka cluster. Unmodified Kafka producers (librdkafka, the Java client, Debezium Server, Kafka Connect, a Materialize Kafka sink) write to it by pointing bootstrap.servers at the replica, with no Kafka cluster in between. The design doc, doc/developer/design/20261030_kafka_listener_source.md, is the place to start.

CREATE SOURCE cdc IN CLUSTER ingest
  FROM KAFKA LISTENER (TOPIC PREFIX 'cdc.public.', SASL USERNAME 'u', SASL PASSWORD SECRET s);
CREATE TABLE orders FROM SOURCE cdc (REFERENCE "cdc.public.orders")
  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr ENVELOPE DEBEZIUM;

How it works

• The replica is the broker. Each clusterd process runs a Kafka protocol server (mz-kafka-server, on the pure-Rust kafka-protocol crate) on a new kafka port. Ingestion stays on the cluster, scaling and isolated like every other source.
• A log of every record. The source appends every batch it accepts to its log, a persist shard, and acknowledges it once that append commits, in about 0.1 s. Tables decode their topic's records from the log, so a topic can feed several tables, and a table created later is filled from the start of its topic.
• Exactly once for idempotent producers. The broker's idempotence rules run against batch rows kept in the same log append as the records, so a resent batch is acknowledged with its original offsets, across restarts. Other producers get at-least-once delivery.
• The Kafka source's pipeline. The source emits the same SourceMessages a Kafka source would, so formats, envelopes (UPSERT, DEBEZIUM, NONE), INCLUDE metadata and reclocking are shared.
• Also: VALIDATE ON PRODUCE rejects records a table could not decode; a read-only Confluent-compatible schema registry per replica for inline Avro schemas; transactions and compacted progress topics, so a Materialize Kafka sink can write into a listener exactly once; and SNI routing through balancerd for Kubernetes. orchestratord wiring is not done.

Both features are behind flags that default to off: enable_kafka_listener_source and enable_replica_kafka_listener.

Commits

Grouped by area, for review. Lines are added plus deleted. Code includes comments; tests count integration tests, test data, and in-file #[cfg(test)] modules.

#	Commit	Code	Tests
1	Design doc
2	ccsr: allow a base path in schema registry URLs. Independent of the rest, and can land on its own.	6	45
3	mz-kafka-server: the protocol server, idempotence, transactions, validation hook, schema registry	2,714	1,180
4	SQL: CREATE SOURCE .. FROM KAFKA LISTENER, TOPIC PREFIX with CREATE TABLE .. FROM SOURCE, the connection type	765	204
5	Storage: the source, its log and appender, validation on produce, log finalization	2,014	56
6	Controller and clusterd: replica ports, orchestrator endpoints, advertised addresses	517
7	balancerd: Kafka and schema registry routing by SNI	328	114
	Total	6,344	1,599

Not part of a production change:

What	Lines
Design doc (commit 1), reviewed rather than shipped as code	787
Demo and end-to-end scripts with their README in misc/kafka-listener (commit 8), for local runs, not CI	1,602
Cargo.lock, generated	68

Measured

On an AWS m6id.16xlarge with 16 producers sending flat out:

• Plain producers: 976k records/s on 4 workers, 1.71M on 16.
• Idempotent producers: 1.51M records/s, exactly once, on 16 workers.
• Correctness: every run held exactly the acknowledged records.

Writing every record to persist twice costs about twice the replica CPU and memory of acknowledging at the table's commit. The extra is persist's encoding and compaction of the log. The design doc covers this and what it flags for persist.

Tests

• mz-kafka-server integration tests with librdkafka producers: compression, idempotence, SASL, transactions, compacted topics, validation mid-sequence, the schema registry, and the byte-bounded topic queue.
• test/sqllogictest/kafka_listener.slt for the SQL.
• Parser testdata.
• misc/kafka-listener scripts for exactly-once delivery through restarts, tables per topic, the registry, and throughput.

🤖 Generated with Claude Code