Who Needs Kafka? Simple PubSub on Postgres Using LISTEN/NOTIFY

Paul Whalen

whoami

Senior Staff Software Engineer @ PEAK6 Capital Management
- Proprietary options trading firm in Chicago
~10 YOE as primarily-backend application engineer (Java, Go, Python)
- Data engineering experience as well, with lots of Kafka and SQL
Database nerd

Agenda

PubSub in Theory
- On Kafka
- On Postgres, with LISTEN/NOTIFY
LISTEN/NOTIFY in Practice
- Config blob service
- Shopping cart
- Time series server
- Ticker viewer
Summary of patterns and gotchas

PubSub in Theory

What is PubSub?

A messaging pattern
Publishers send messages for a topic
Subscribers receive messages on a topic
Subscribers don’t know about publishers, publishers don’t know about subscribers
- Something in the middle (a “broker”) connects the two
Optionally:
- Usually real time
- Persistent or ephemeral
- “Queue” behavior, marking messages as processed

What is PubSub?

Kafka as an implementation of PubSub

Producers publish messages on a topic to broker
- Messages are written in order, and assigned IDs (“offset”)
Consumers subscribe to a topic and recieve messages from broker
- Read from an offset

Kafka key points

Messages are persisted
Consumer can read older messages
Consumers process every message
Consumers read the entire message

Postgres LISTEN/NOTIFY

NOTIFY is the “publish” command (writer)
LISTEN is the “subscribe” command (reader)
NOTIFY sends a message on a channel
Any connection currently LISTENing on that channel receives the message

Publisher:

SELECT pg_notify('a_channel', 'hello world');

Subscriber:

LISTEN a_channel;

“Kafka” on Postgres

A table per topic
Publisher writes to table, NOTIFY ID
Susbcriber LISTEN, reads messages from table

“Kafka” on Postgres

Table to store messages

CREATE TABLE a_topic (
    id SERIAL PRIMARY KEY,
    message TEXT
)

Subscriber listens

LISTEN a_topic;

Publisher writes message…

INSERT INTO a_topic (message)
VALUES ('hi')
RETURNING id

… and notifies that it has been written (transactionally)

SELECT pg_notify('a_topic', '1')

“Kafka” on Postgres

Subscriber receives notification…

'1'

… and reads message

SELECT * FROM a_topic WHERE id = 1

id	message
1	hi

LISTEN/NOTIFY in Production

Config Blob Service

A service that provides create/read/update/delete storage for a blob
Updates are pushed out immediately to the front end
The configuration is opaque to the DB, understood only by the application

CREATE TABLE profile_config
(
    profile_id  SERIAL PRIMARY KEY,
    modified_at TIMESTAMPTZ DEFAULT now(),
    config      JSON
)

Writing Data

Example config blob

SELECT json_build_object(
    'profile_id', 1,
    'config_key', 'config_val'
) AS data

data
{‘profile_id’: 1, ‘config_key’: ‘config_val’}

Writing Data

Insert blob

INSERT INTO profile_config (modified_at, config)
VALUES (
    now(),
    json_build_object(
        'profile_id', 1,
        'config_key', 'config_val'
    )
)

Notify config_update channel with contents of blob

SELECT pg_notify(
    'config_update',
    json_build_object(
        'profile_id', 1,
        'config_key', 'config_val'
    )::text
)

Reading Data

Subscriber LISTENs (before data was written)

LISTEN config_update;

Receives

{‘profile_id’: 1, ‘config_key’: ‘config_val’}

… there’s no SELECT!

Cool!

Faster
Less indirection
Like Kafka: only receive, not the notify then read

Large write

But what if the newly written config blob is big?

SELECT pg_notify(
    'config_update',
    json_build_object(
        'profile_id', 1,
        'config_key', REPEAT('a', 10000)
    )::text
)

---------------------------------------------------------------------------
InvalidParameterValue                     Traceback (most recent call last)
Cell In[10], line 1
----> 1 run_sql("""
      2 SELECT pg_notify(
      3    'config_update',
      4    json_build_object(
      5        'profile_id', 1,
      6        'config_key', REPEAT('a', 10000)
      7    )::text
      8 )
      9 """, print_sql=True, print_output=False)

Cell In[1], line 34, in run_sql(sql, print_sql, print_output, client)
     32 cursor = conn.cursor()
     33 cursor.execute("SET schema 'pubsub'")
---> 34 cursor.execute(sql)
     35 if print_output:
     36     fetch_and_print_md_table(cursor)

InvalidParameterValue: payload string too long

Large write problems

❗ Gotcha ❗: NOTIFY payloads have a hard 8kb limit
You must know the data you’re putting in the payload
- Don’t accept untrusted data
- In our case, it wasn’t malicious, and it was only just barely over

Large writes

Kafka has a message limit size too, but it is much larger (1MB)
- (with config, you can get it up over 100MB)
I have also been the idiot writing large messages in Kafka
In Postgres, you can decouple the notification from the read/write
🟢 Pattern 🟢: NOTIFY instructions about what to read, not the data itself
- Many times, this is the ID of a row, but not always…

Shopping Cart

A service that manages an online shopping cart (e.g. Amazon)
When an item is added to the cart, a row is added to the table
A front end is connecting to one instance of the server that is LISTENing to postgres to show the updated cart
- There are multiple instances of the service

Shopping Cart Schema

CREATE TABLE shopping_cart
(
    id       SERIAL PRIMARY KEY,
    item     TEXT NOT NULL,
    username TEXT NOT NULL
)

Adding an item to the cart

Add item

INSERT INTO shopping_cart (item, username)
VALUES ('iPhone', 'paul')

Rather than NOTIFYing the ID, we NOTIFY for the user:

SELECT pg_notify('shopping_cart', 'paul')

🟢 Pattern 🟢: NOTIFY instructions about what to read, not the data itself

Reading the updated cart

Listening

LISTEN shopping_cart;  -- 'paul'

Read the entire cart for the user

SELECT *
FROM shopping_cart
WHERE username = 'paul'

id	item	username
1	iPhone	paul

Querying for the whole cart

Querying for the entire shopping cart is easiest
- Service doesn’t need to track to cart, it just refreshes it
- The whole cart is never prohibitively large
With Kafka, you get all the messages
- Must track the state of the cart in the service
🟢 Pattern 🟢: On LISTEN, query for the whole state
- Let the database manage the state
Stateful stream processing with Kafka:
- The greatest mistake of my career

Not querying at all

A given user front end is only connected to one instance
We only need to read the cart if the user is connected to that instance
With Kafka, you have to receive the contents of all messages
- Exception: partitions
- Partitions can be clumsly; you can only partition one way

Bonus: Handling deletes

Add a column representing items that were removed from the cart

ALTER TABLE shopping_cart
    ADD COLUMN removed BOOLEAN DEFAULT FALSE

User removes from cart

UPDATE shopping_cart
SET removed = true
WHERE ID = 1

SELECT pg_notify('shopping_cart', 'paul')

SELECT *
FROM shopping_cart
WHERE username = 'paul'
AND NOT removed

… there’s nothing in the cart!

🟢 Pattern 🟢: Query for the whole state, let the database manage the state

Deletes in Kafka

“Tombstone” messages indicating delete
Application must update state to handle delete
- Often surprisingly hard, especially when it involves outputting more tombstone messages

Time Series Server

Time series storage of stock prices throughout the day
Traders want to see up-to-date aggregations, like min/max/avg

CREATE TABLE stock_price
(
    ticker    TEXT,
    timestamp TIMESTAMPTZ,
    price     NUMERIC
)

A write-heavy workload

Stock price changes a lot

INSERT INTO stock_price (ticker, timestamp, price)
VALUES ('AAPL', '2025-01-01 8:30:00', 100.00),
       ('AAPL', '2025-01-01 8:30:01', 100.01),
       ('AAPL', '2025-01-01 8:30:02', 100.02)

SELECT pg_notify('stock_price', 'AAPL');
SELECT pg_notify('stock_price', 'AAPL');
SELECT pg_notify('stock_price', 'AAPL')

LISTEN stock_price;

'AAPL'
'AAPL'
'AAPL'

Serving up-to-date aggregations

On LISTEN, we do this aggregation query 3 times

SELECT ticker, avg(price) FROM stock_price
GROUP BY ticker

ticker	avg
AAPL	100.0100000000000000

This query could be intensive with a lot of data
Queries could get backed up behind each other on thread the service is using to LISTEN
❗ Gotcha ❗: The LISTEN thread can get backed up
🟢 Pattern 🟢: Do your work elsewhere from the LISTEN thread
If stock prices are changing faster than the user can see anyway, do you need to keep up?
🟢 Pattern 🟢: LISTENer doesn’t need to care about every message

Ticker View

Traders are constantly switching between stock tickers to view data related to a given ticker
When a trader switches what ticker they’re viewing, each UI receives a message from the service with the ticker to load data for

Ticker View Architecture

Ticker View Schema

CREATE TABLE ticker_view
(
    id        SERIAL PRIMARY KEY,
    username  TEXT NOT NULL,
    ticker    TEXT NOT NULL
)

Basic Ticker View Pub Sub

We’ve seen all this before…

INSERT INTO ticker_view (username, ticker)
VALUES ('paul', 'AAPL')
RETURNING id

SELECT pg_notify('ticker_view', '1')

LISTEN ticker_view;

SELECT ticker FROM ticker_view
WHERE id = 1

ticker
AAPL

Goal: High Availabilitiy

What if the UI briefly loses connection to the service?
What if the service briefly loses connection to the database?
❗ Gotcha ❗: You miss messages when you aren’t connected

Querying for what we missed

How do we know what we missed?
Let’s assume the UI was keeping track of the ID of the last message it received
- In this case, 0 to indicate it hasn’t received any messages

SELECT ticker FROM ticker_view
WHERE id > 0
AND username = 'paul'

ticker
AAPL

Multiple missed messages, like Kafka

But what if we missed multiple messages?

SELECT id, ticker FROM ticker_view
WHERE id > 0
AND username = 'paul'

id	ticker
1	AAPL
2	NVDA

Querying since an ID is analogous to a concept in Kafka called consumer groups
Consumer Groups help you pick up where you left off, by keeping track of the last message ID (“offset”)
With Kafka you have to consume every message, which we don’t want to do in this case

Multiple missed messages, with Postgres

For this use case, we only care about the last message of the ones that we missed

SELECT id, ticker FROM ticker_view
WHERE username = 'paul'
AND id > 0
ORDER BY id DESC LIMIT 1;

id	ticker
2	NVDA

This is more useful to us than Kafka Consumer Groups!
Not having to read every message is an advantage
It all depends on the use case
🟢 Pattern 🟢: LISTENer doesn’t need to care about every message

Patterns and Gotchas Summary

Patterns and Gotchas

❗ Gotcha ❗: NOTIFY payloads have a hard 8kb limit
🟢 Pattern 🟢: NOTIFY instructions about what to read, not the data itself
🟢 Pattern 🟢: LISTENer doesn’t need to care about every message
🟢 Pattern 🟢: On LISTEN, query for the whole state, let the database manage the state
❗ Gotcha ❗: The LISTEN thread can get backed up
🟢 Pattern 🟢: Do your work elsewhere from the LISTEN thread

Some notes on scale, Kafka, and Postgres

Kafka is good for lots of messages and lots of subscribers
Do you really need to send that many messages?
Do you really need that many subscribers?
You can get far on Postgres using the right patterns

Thanks For Listening!

Questions?

Presentation: pgwhalen.com/pg-pubsub-talk
Source: github.com/pgwhalen/pg-pubsub-talk/