gem-hunter

A PostgreSQL extension that embeds a Kafka broker. Unmodified Kafka clients (librdkafka, the Java client, Sarama, kafka-python, kcat, the kafka-*.sh tooling) connect to Postgres on port 9092 and cannot tell the difference. Topics, partitions, offsets, consumer groups, and the log itself live in the database.

The goal is transactional coupling between the database and the event stream:

BEGIN;
  INSERT INTO orders (id, customer_id, total) VALUES (...);
  SELECT kafgres_produce('order-events', 'OrderCreated',
                         jsonb_build_object(...)::text);
COMMIT;

Both commit or neither does. There is no outbox table, no Debezium, no Connect cluster, and no dual-write inconsistency window. See docs/producing.md for the produce paths and when to use each.

The following all work today:

• Unmodified clients produce, consume, and rebalance: librdkafka and kcat, the Java client and its console tools, Sarama, and kafka-python.
• A default-config Java KafkaProducer with idempotence on (the default since Kafka 3.0) works with no overrides.
• Admin APIs: kafka-topics.sh, kafka-configs.sh, kafka-consumer-groups.sh, kafka-acls.sh, transactions, share groups, and the KIP-848 consumer protocol.
• Retention reclaims disk, and cleanup.policy=compact topics are compacted on both storage engines.
• Clients authenticate with SCRAM-SHA-256 against Postgres roles or an mTLS certificate, with ACLs in a SQL-managed table.
• Consumers truncate rather than read divergent data when an async replica is promoted, verified against a real physical standby.
• The conformance suite drives five real clients (Java, librdkafka, Sarama, franz-go, kafka-python) through some 180 scenarios against kafgres and a reference Kafka and diffs the observable results; it runs in CI on every pull request, and every intended difference is catalogued in docs/conformance.md.
• The segment engine (the default) passes the same conformance suite, survives kill -9 with every acknowledged record intact, and replicates its log to a standby out of band. By default every acknowledged record also survives a host crash. On an i9-13900 with power-loss-protected NVMe it produced about 2.4x the table engine's throughput with both durability settings relaxed. The strict defaults cost 3 to 25%: 806 MB/s relaxed against 605 MB/s strict for 1 KiB records from 4 producers (the full table is in docs/configuration.md).
• kafgres_produce() commits atomically with a business write.
• CDC: a table's changes reach a topic through a logical decoding output plugin shipped with the extension, with the mapping written in SQL. See docs/producing.md.

If you run the segment engine, set kafgres.segment_archive_command. The segment engine's log lives in files, so pg_basebackup seeds a replica but is not a backup: retention unlinks rolled segments that an archive would still want. The setting takes a shell command with %p/%f, exactly as Postgres's own archive_command does, and retention refuses to reclaim a segment the archive has not taken. As with WAL archiving, a failing command stops reclamation and the disk grows, so watch kafgres_archive_status(). The table engine needs none of this; its log is in Postgres tables and pg_basebackup already covers it.

docker compose up -d starts a Postgres preloaded with kafgres and creates the extension on first start (CREATE EXTENSION kafgres runs from the image's init scripts). On a cluster you administer yourself, add kafgres to shared_preload_libraries and run CREATE EXTENSION kafgres;.

demo/kcat-demo.sh then points kcat at it and runs the commands you would run against any Kafka broker. It needs the broker up; with no local kcat it builds the small test-client image on first use:

docker compose up -d
bash demo/kcat-demo.sh

kafgres: a Kafka broker inside PostgreSQL

There is no broker process. Port 9092 is served by a Postgres background worker:

$ psql -tAc "SELECT 'PostgreSQL ' || current_setting('server_version')"
  PostgreSQL 16.14 (Debian 16.14-1.pgdg12+1)

$ psql -tAc "SELECT backend_type FROM pg_stat_activity
                   WHERE backend_type = 'kafgres_broker'"
  kafgres_broker

Topics are created in SQL, because a topic is a row:

$ psql -tAc "SELECT kafgres_create_topic('payments', 3)"
  121

Everything from here is plain kcat, pointed at 127.0.0.1:9092.

$ kcat -b 127.0.0.1:9092 -L -t payments
  Metadata for payments (from broker 1: 127.0.0.1:9092/1):
   1 brokers:
    broker 1 at 127.0.0.1:9092 (controller)
   1 topics:
    topic "payments" with 3 partitions:
      partition 0, leader 1, replicas: 1, isrs: 1
      partition 1, leader 1, replicas: 1, isrs: 1
      partition 2, leader 1, replicas: 1, isrs: 1

Produce three keyed records. The client hashes the key to a partition, so this
also checks that the broker serves the partition the client chose:

$ printf 'erin:{"amt":120}\nken:{"amt":80}\nalice:{"amt":210}\n' \
    | kcat -b 127.0.0.1:9092 -t payments -K: -P

Read them back with partition, offset and key:

$ kcat -b 127.0.0.1:9092 -t payments -C -e -q -o beginning -f '%p:%o  %k -> %s\n'
  0:0  erin -> {"amt":120}
  1:0  ken -> {"amt":80}
  2:0  alice -> {"amt":210}

Consumer groups work. The second run starts from the committed offset and
finds nothing left:

$ kcat -b 127.0.0.1:9092 -G payments-demo payments -e -q -o beginning -c 3 -f '%p:%o  %k -> %s\n'
  1:0  ken -> {"amt":80}
  0:0  erin -> {"amt":120}
  2:0  alice -> {"amt":210}

$ kcat -b 127.0.0.1:9092 -G payments-demo payments -e -q -f 'unexpected: %p:%o\n'
  (no output)

The log is part of the same database as the application tables that produced
them — here are its offsets, straight from SQL:

$ psql -c "SELECT partition, COALESCE(high_watermark, 0) AS log_end_offset,
                log_start_offset FROM kafgres_partition_offsets('payments')"
   partition | log_end_offset | log_start_offset
  -----------+----------------+------------------
           0 |              1 |                0
           1 |              1 |                0
           2 |              1 |                0

The script uses a local kcat if you have one and the test client image otherwise. Nothing in it is kafgres-aware: every kcat line is one you would run against a real broker.

Install the new build, restart Postgres, then run ALTER EXTENSION kafgres UPDATE; in the database the broker uses. The broker serves across the restart before the update runs; the update changes who may call the administrative functions (see docs/producing.md) and pins kafgres_produce()'s search path.

From 0.2.0, the first start on the segment engine removes the log's old-format .timeindex files. Nothing is lost: timestamp lookups scan the segments written before the upgrade instead, and segments written after it are indexed.

From 0.3.0, transaction markers written before 0.3.1 carry no EndTransactionMarker value, which Sarama fails to decode. They stay until retention removes their segments. A Sarama consumer can start past the last such marker, or the topic can be recreated.

A table's changes reach a topic without Kafka Connect, a second JVM, or a network hop. The mapping is SQL, not a template, so it can join, filter with a real predicate, and build jsonb that is typed by construction:

CREATE TABLE customers (id int PRIMARY KEY, name text, tier text);
CREATE TABLE orders (
    id          int PRIMARY KEY,
    customer_id int REFERENCES customers,
    total       numeric,
    status      text
);
INSERT INTO customers VALUES (1, 'Acme', 'gold');

SELECT kafgres_create_topic('order-events', 3);

SELECT kafgres_add_mapping(
    'orders', 'public.orders', 'order-events',
    value_expr  => $$ jsonb_build_object(
                        'order_id', new.id,
                        'total',    new.total,
                        'op',       op,
                        'customer', (SELECT jsonb_build_object('tier', c.tier)
                                       FROM customers c WHERE c.id = new.customer_id)) $$,
    key_expr    => $$ new.id::text $$,
    filter_expr => $$ new.status = 'placed' $$);

A background worker drains a logical replication slot fed by kafgres's own output plugin, renders each mapping, and produces:

$ psql -c "INSERT INTO orders VALUES (10, 1, 99.50, 'placed')"
$ kcat -b 127.0.0.1:9092 -t order-events -C -e -q -o beginning -f '%k -> %s\n'
  10 -> {"op": "I", "total": 99.50, "customer": {"tier": "gold"}, "order_id": 10}

old is in scope too, and comes out of the WAL rather than a re-read, so on a table with REPLICA IDENTITY FULL the before-image is exact as of the change. See docs/producing.md for the full mapping interface.

One caveat: an enrichment subquery runs when the change is rendered, so it reads current state rather than the state the transaction saw. That is the same trade a Debezium SMT lookup or a downstream stream-join makes. When an event must reflect exactly what the transaction saw, use kafgres_produce(), which runs inside the transaction.

demo/parity-demo.sh runs the same commands against kafgres and against a real apache/kafka broker and diffs the output. It needs both images and nothing else:

docker compose build && docker build -t kafgres-clients tests/clients
bash demo/parity-demo.sh

kafgres parity demo
Left:  kafgres: a Kafka broker inside PostgreSQL, on 127.0.0.1:9092
Right: apache/kafka: a reference Kafka broker, on 127.0.0.1:9292
Every check runs the identical command against both and diffs the output.

Admin tooling (the Java client, unmodified)
  ✓ kafka-topics.sh --describe                     identical
  ✓ kafka-topics.sh --list                         identical
  ✓ kafka-configs.sh --describe                    identical
  ✓ kafka-get-offsets.sh                           identical

Produce and consume (librdkafka / kcat)
  ✓ kcat -P then -C, 6 keyed records over 3 partitions identical
  ✓ kcat -L metadata                               identical

Independent clients (no shared code with librdkafka or Java)
  ✓ Sarama (Go) - produce, consume, offsets        identical
  ✓ Sarama (Go) - consumer group, commit           identical
  ✓ kafka-python - produce, consume, offsets       identical
  ✓ kafka-python - consumer group, commit          identical
  ✓ unknown topic returns the same error code      identical

Consumer groups
  ✓ kafka-consumer-groups.sh --describe (lag, offsets) identical
  ✓ kafka-consumer-groups.sh --list                identical

The difference
Everything above matched. This does not:

  $ psql -c "SELECT partition, COALESCE(high_watermark, 0) AS log_end_offset,
                log_start_offset FROM kafgres_partition_offsets('parity-demo')"
   partition | log_end_offset | log_start_offset
  -----------+----------------+------------------
           0 |              2 |                0
           1 |              1 |                0
           2 |              3 |                0

13/13 checks identical  ✓

A check that fails prints the diff and the script exits non-zero. Two differences are normalized away, both catalogued in docs/conformance.md: min.insync.replicas, which the broker does not report because it does not honour it, and the ELR columns, which the tool renders as N/A against kafgres because it falls back to Metadata. Anything else that differs is treated as a bug, and the script shows it rather than absorbing it.

The demo is a summary; tests/conformance/ is the gate, with the five clients run through about 180 scenarios against both brokers in CI on every pull request.

The Kafka protocol has been reimplemented several times over non-Kafka logs: Redpanda (C++), WarpStream (S3), AutoMQ (S3-backed fork), Kafka-on-Pulsar. kafgres differs by using Postgres as the log and by producing inside the same transaction as the application write.

codec/          kafgres-codec, the wire protocol. No pgrx, no Postgres, unit-testable alone.
  schemas/      Kafka message schemas, vendored at a pinned tag (codec/VENDORING.md)
  implemented.toml   which APIs are served, and at which versions
  src/generated/     emitted by codec-gen and checked in; do not edit
codec-gen/      the generator. `cargo run -p kafgres-codec-gen`
extension/      the pgrx extension. Its own workspace on purpose.
docs/           architecture.md, producing.md, conformance.md, configuration.md

cargo test                    # codec + generator. No Postgres needed.
docker compose build          # links the extension; cargo check does not
docker compose up -d

docker build -t kafgres-clients tests/clients
pip install -r requirements.txt   # pytest + kafka-python for the test suites
pytest tests/integration/         # real clients against a running broker

Try it:

SELECT kafgres_create_topic('orders', 3);

$ kcat -b localhost:9092 -L
 1 brokers:
  broker 1 at localhost:9092 (controller)
  topic "orders" with 3 partitions:
    partition 0, leader 1, replicas: 1, isrs: 1

cargo check does not exercise the pgrx/Postgres link step and passes on code that cannot load. Use docker compose build.

The image builds against PostgreSQL 16 by default; set PG_MAJOR (13 through 18) to build against another major, as in PG_MAJOR=18 docker compose build. CI builds and tests every major from 13 to 18.

Released under the Elastic License 2.0; see LICENSE. The Kafka message schemas vendored under codec/schemas/ are Apache License 2.0, upstream Apache Kafka; see codec/VENDORING.md.