Who Needs Kafka? PubSub Patterns Beyond Message Queues on Postgres Using LISTEN/NOTIFY
Paul Whalen
whoami
- Senior Staff Software Engineer @ PEAK6 Capital Management
- Proprietary options trading firm in Chicago
- ~11 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
NOTIFYis the “publish” command (writer)LISTENis the “subscribe” command (reader)NOTIFYsends a message on a channel- Any connection currently
LISTENing on that channel receives the message
“Kafka” on Postgres
- A table per topic containing the messages
- Publisher writes to table,
NOTIFYID - Susbcriber
LISTENs, reads messages from table
“Kafka” on Postgres
Table to store messages
Subscriber listens
Publisher writes message…
… and notifies that it has been written (transactionally)
“Kafka” on Postgres
Subscriber receives notification…
'1'… and reads message
| id | message |
|---|---|
| 1 | hi |
A “queue” on Postgres
Table to store messages
CREATE TABLE a_queue (
id SERIAL PRIMARY KEY,
message TEXT,
processed BOOLEAN DEFAULT FALSE -- new column
)Subscriber listens
Publisher writes message…
… and notifies that it has been written (transactionally)
A “queue” on Postgres
LISTEN/NOTIFY Beyond Topics and Queues, In Production
Config Blob Service
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
Writing Data
Example config blob
| 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
Reading Data
Subscriber LISTENs (before data was written)
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[14], 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 40, in run_sql(sql, print_sql, print_output, client) 38 cursor = conn.cursor() 39 cursor.execute("SET schema 'pubsub'") ---> 40 cursor.execute(sql) 41 if print_output: 42 fetch_and_print_md_table(cursor) InvalidParameterValue: payload string too long
Large write problems
- ❗ Gotcha ❗:
NOTIFYpayloads 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 🟢:
NOTIFYinstructions about what to read, not the data itself- Many times, this is the ID of a row
Shopping Cart
Shopping
- 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 and is
LISTENing to postgres- There are multiple instances of the service
Adding an item to the cart
Rather than NOTIFYing the ID, we NOTIFY for the user:
🟢 Pattern 🟢: NOTIFY instructions about what to read, not the data itself
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 (soft) deletes
Add a column representing items that were removed from the cart
User removes from 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 Server
- Time series storage of stock prices throughout the day
- Traders want to see up-to-date aggregations, like min/max/avg
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')'AAPL'
'AAPL'
'AAPL'Serving up-to-date aggregations
On LISTEN, we do this aggregation query 3 times
- 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
LISTENthread can get backed up - 🟢 Pattern 🟢: Do your work elsewhere from the
LISTENthread - 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
Alternatively: Batch Writes
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)Only notify once for the whole batch
- 🟢 Pattern 🟢:
LISTENer doesn’t need to care about every row - Disadvantage: increased latency while batch accumulates
Ticker View
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
Basic Ticker View Pub Sub
We’ve seen all this before…
| 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,
0to indicate it hasn’t received any messages
- In this case,
Multiple missed messages, like Kafka
But what if we missed multiple messages?
| id | ticker |
|---|---|
| 1 | AAPL |
| 2 | NVDA |
- This 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
| 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 ❗:
NOTIFYpayloads have a hard 8kb limit - 🟢 Pattern 🟢:
NOTIFYinstructions 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
LISTENthread can get backed up - 🟢 Pattern 🟢: Do your work elsewhere from the
LISTENthread
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/
Who Needs Kafka? PubSub Patterns Beyond Message Queues on Postgres Using LISTEN/NOTIFY Paul Whalen