Kafgres: Embedding a Kafka Broker into Postgres

TL;DR

“Just use Postgres”

• Mostly compelling
• But why throw away the Kafka ecosystem

Dozens of blogs, listicles and wikis center around the concept of “just use Postgres’. Many have contrasted the operational burden and cost of managing a Kafka cluster against just using various techniques to get message queues or event logs out of your existing Postgres clusters.

Proposed options range from using Postgres locks to handle competitive consumption from tables (that can be treated as logs).

SELECT FOR UPDATE SKIP LOCKED

To using purpose built extensions extending psql for message queues

SELECT * from pgmq.send(
  queue_name  => 'my_queue',
  msg         => '{"foo": "bar1"}'
);

These blog posts are broadly correct, in that Postgres (especially when extended) is sufficient for the vast majority of use-cases, especially at lower scales, and that relying on your existing database alleviates a large operational burden.

One caveat not usually addressed is that relying on bespoke table-row-based queues or tailor-made Postgres extensions throws away the Kafka ecosystem, and leaves you with a smaller corpus of libraries to draw on during development. In many cases, that is a reasonable trade-off, but “writing wrappers around Postgres” is not most people’s end goal.

I have been working on kafgres, an extension embedding a performant singleton Kafka broker into Postgres. Not only does doing this let you reuse your database as your log broker and store, it allows tight coupling between your event log and database activity, without needing to spin up yet another deployment (i.e. debezium).

$ psql -h Postgres -p 5432 -U Postgres -c "SELECT version()"
PostgreSQL 16.15 (Debian 16.15-1.pgdg12+2) on aarch64-unknown-linux-gnu, compiled by gcc (Debian 12.2.0-14+deb12u1) 12.2.0, 64-bit

$ kcat -b Postgres:9092 -L
Metadata for all topics (from broker 1: Postgres:9092/1):
 1 brokers:
  broker 1 at Postgres:9092 (controller)
BEGIN;
  INSERT INTO orders (id, customer_id, total, status)
  VALUES (1001, 1, 249.99, 'placed');

  SELECT kafgres_produce(
      'order-events',
      'order-1001',
      jsonb_build_object(
          'event',    'OrderPlaced',
          'order_id', 1001,
          'customer', 'Acme Corp',
          'total',    249.99
      )::text) AS offset_assigned;
COMMIT;

Of course, Postgres already provides its own WAL, which is ultimately what debezium usually reads. If you’re more used to mapping events from the CDC than emitting via Postgres, you can do that as well.

SELECT kafgres_add_mapping(
    'orders-cdc',                       -- mapping name
    'public.orders',                    -- source table
    'orders-cdc',                       -- destination topic
    value_expr  => $$ jsonb_build_object(
                        'op',       op,
                        'order_id', new.id,
                        'total',    new.total,
                        'status',   new.status,
                        'customer', (SELECT jsonb_build_object('name', c.name, 'tier', c.tier)
                                       FROM customers c WHERE c.id = new.customer_id)) $$,
    key_expr    => $$ new.id::text $$,
    filter_expr => $$ new.status <> 'draft' $$);

Kafgres isn’t just a server pointed at your tables (doing this would be prone to causing WAL churn), it is built on a pluggable storage engine living in the Postgres engine, and it implements dozens of Kafka APIs, so that both Kafka clients and admin tools sees it as a regular, singleton Kafka broker.

$ echo 'order-3001:{"event":"OrderPlaced","order_id":3001,"source":"Kafka-client"}' \
    | kcat -b Postgres:9092 -t order-events -K: -P

$ psql -h Postgres -c "BEGIN;
    INSERT INTO orders VALUES (3002, 4, 1250.00, 'placed');
    SELECT kafgres_produce('order-events', 'order-3002',
        jsonb_build_object('event','OrderPlaced','order_id',3002,'source','sql-transaction')::text);
  COMMIT;"

$ kcat -b Postgres:9092 -t order-events -C -e -q -o beginning -f '%p:%o  %k -> %s\n'
  0:0  order-1001 -> {"event": "OrderPlaced", "total": 249.99, "customer": "Acme Corp", "order_id": 1001}
  1:0  order-3001 -> {"event":"OrderPlaced","order_id":3001,"source":"Kafka-client"}
  1:1  order-3002 -> {"event": "OrderPlaced", "source": "sql-transaction", "order_id": 3002}

On a machine with an NVME and an i7 from 2017, a Postgres instance with kafgres installed is stable and fluid at a throughput of 30k-70k event/s. The DB remains responsive during this time, and with the default storage engine, we only see a few percent-point hit in pgbench performance.

Kafgres versus:

In an order pipeline demo, we are able to reduce a 3-deployment event-driven architecture to one Postgres deployment, while still using standard Kafka libraries to build our microservices.

I have recently released Kafgres 0.1.0, and will be continuing to iterate on the broker to chase down performance, safety, and utility gains.

Misc. on the writing of Kafgres

Kafgres is written using pgrx, a framework that I am a big fan of, allowing you to write Postgres extensions in Rust. While performance is important for the extension, considering we want a performant broker, the main appeal of Rust for this project is reducing the footprint of code that could crash the broker. If Kafgres were super prone to crashing, the appeal of collapsing two deployments to one would be defeated by the risk of bringing down your one Postgres deploy.

I transported a lot of learnings from my previous broker-in-Postgres project, pgmqtt. Where we diverge from the current version of pgmqtt is leaving a lot more internals off of the WAL. Although this means we need to write a lot more code, it means we can achieve a much higher throughput without crowding the other Postgres workers.

Q: If we keep our topic off the WAL, how do we get transactional kafgres_produce()?

A: kafgres_produce uses a commit marker row that does land on the WAL, and is inserted in the caller's transaction. The produced message doesn't land in that transaction, but the row that tells the kafgres engine whether the message commits does. This allows us to replicate Kafka's own aborted-transaction protocol.

Q: If Postgres is a singleton, how do we handle replication?

A: Take cues from Postgres. We stream replication data to a standby (another kafgres, in another Postgres). Since we avoid using the WAL for topic data, the kafgres follower needs to pull the log using the Kafka fetch protocol (as if it were reading from another, regular Kafka broker). We replicate Kafka's epoch reconciliation to handle divergent data.