DEV Community: Eric Goldman

How PostgreSQL logical decoding and plugins work

Eric Goldman — Thu, 22 May 2025 20:24:00 +0000

When you're building a change data capture (CDC) pipeline with PostgreSQL, one of the first decisions you'll make is which output plugin to use. These plugins determine how your database changes get formatted and delivered—whether you're replicating to another Postgres instance, streaming to Kafka, or delivering changes to a webhook.

But what exactly are these plugins, and how do they differ in practice? Let's dive in with a concrete example and explore your options.

A simple example: see an update flow end-to-end

To understand how different output plugins work, let's start with a simple scenario. We'll set up logical replication, make a change, and see how different plugins format that change.

Setting up logical replication

First, you need to enable logical replication in PostgreSQL. You can read the Sequin docs to get specific instructions for Postgres providers like AWS, GCP, and Azure. But at a high level, you can run the command show wal_level; . If it’s not, you can follow these steps to enable it.

Create a table and replication slot

Create a table (if you don’t have one already) and create a publication and replication slot:

-- 1. Create a test table
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    email VARCHAR(255) NOT NULL,
    created_at TIMESTAMP DEFAULT NOW()
);

-- 2. Set the replica identity to full for your table so all changes flow through:
alter table public.properties replica identity full;

-- 2. Insert initial data
INSERT INTO users (name, email) VALUES ('John Doe', 'john@example.com');

-- 3. Crate a publication
create publication cool_pub_name for all tables with (publish_via_partition_root = true);

-- 4. Create a logical replication slot with your chosen plugin
-- Start with test_decoding to get familiar:
SELECT pg_create_logical_replication_slot('test_slot', 'test_decoding');

Now make the change we want to track:

-- The change we're tracking
UPDATE users SET name = 'John Smith' WHERE id = 1;

Reading changes from the replication slot

To see the formatted output, you can read from your replication slot:

-- For test_decoding (simplest to read): 
SELECT lsn, xid, data FROM pg_logical_slot_peek_changes('test_slot', NULL, NULL);

This simple update will look very diff erent depending on which output plugin you're using.

How changes Fflow from table to output plugin

When you executed that UPDATE statement, here's the journey your change took through PostgreSQL to arrive in the replication slot:

1. Write-Ahead Log (WAL)

PostgreSQL's WAL does not store pre-decoded, human-readable messages. Instead, it logs low-level binary records describing each change at the storage level. When our user name update happens, the WAL record looks something like this:

# What's actually in the WAL (simplified)
WAL Record: LSN 0/1A2B3C4
- Relation OID: 16384 (internal table identifier)
- Transaction ID: 12345
- Operation: UPDATE
- Block/offset: physical storage location
- Old tuple: [binary data for old row]
- New tuple: [binary data for new row]

At this stage the WAL only knows about internal identifiers and bytes – it does not contain table names, column names, or data in text form. Any readable output you see (e.g. JSON or text changes) is generated later by the decoding plugin – not stored in the WAL itself.

2. Logical decoding process

A logical replication slot streams changes by decoding WAL records on demand as a client consumes the slot. Here's what happens when you read from your slot:

PostgreSQL starts from your slot's position (LSN) and reads WAL records sequentially
Looks up table metadata: The decoding process uses the WAL record's metadata (like the relation OID) to look up the corresponding table and column definitions in the system catalogs
Transforms internal data: The binary tuple data gets converted into logical representation with actual table names, column names, and typed values
Assembles complete transactions: Internally, the logical decoder assembles transactions in commit order; only after a transaction is fully decoded (and committed) will its changes be passed to the plugin

Importantly, this decoding happens at read time, not when the WAL is originally written.

3. Plugin-specific formatting

The output plugin takes the decoded change data and formats it according to its own rules. For our user update, every plugin receives the same structured information:

Table name: public.users
Operation: UPDATE
New values: {id: 1, name: "John Smith", email: "john@example.com"}
Old values: {name: "John Doe"} (what changed)

Every logical decoding plugin receives the same core information about the change; what differs is how they output it. The test_decoding plugin formats this as human-readable text, wal2json converts it to JSON, and pgoutput encodes it in PostgreSQL's binary logical replication protocol.

This architecture allows PostgreSQL to support many output formats without changing the underlying WAL format or storing duplicate information. The core database only needs to log changes once in the WAL, and then any number of output plugins can decode those logs and present the data in JSON, SQL, binary, etc., as needed.

Built-in output plugins

PostgreSQL ships with two logical decoding plugins out of the box. These don't require any additional installations—they're ready to use on any Postgres 10+ server.

pgoutput

pgoutput is PostgreSQL's default plugin for logical replication. If you're using the built-in publish/subscribe system, you're already using this plugin behind the scenes.

What our user name update looks like:

# Binary protocol message (conceptual representation)
BEGIN LSN: 0/1A2B3C4
TABLE: public.users
UPDATE: id[integer]=1 name[text]='John Smith' (old: 'John Doe') email[text]='john@example.com'
COMMIT LSN: 0/1A2B3C4

The actual output is a binary protocol, so you can't just read it directly. Tools like Sequin or custom replication clients parse this format.

Trade-offs:

✅ Binary format is efficient and compact
✅ Handles complex PostgreSQL data types without losing information
✅ High performance with incremental streaming
✅ Production-ready and universally supported on managed services
✅ Used by PostgreSQL's native logical replication
❌ Requires special tools to read (can't just use SQL functions)
❌ Binary protocol is not human-readable for debugging

test_decoding

test_decoding is PostgreSQL's example plugin. It's mainly useful for understanding how logical decoding works or for quick debugging.

What our user name update looks like:

BEGIN 12345
table public.users: UPDATE: id[integer]:1 name[text]:'John Smith' email[text]:'john@example.com'
COMMIT 12345

Trade-offs:

✅ Text format is human-readable and easy to understand
✅ Great for learning how logical decoding works
✅ Included with PostgreSQL by default
✅ Useful for quick debugging of replication issues
❌ Output format is not designed for production parsing
❌ No advanced features like filtering or sophisticated type handling
❌ Limited functionality compared to other plugins

Popular third-party plugins

The PostgreSQL community has created several plugins for specific integration needs. Here's a high-level overview of the most common ones:

wal2json

wal2json outputs changes in JSON format, making it easy to work with in any programming language.

What our user name update looks like:

{
  "change": [
    {
      "kind": "update",
      "schema": "public",
      "table": "users",
      "columnnames": ["id", "name", "email", "created_at"],
      "columnvalues": [1, "John Smith", "john@example.com", "2024-01-15T10:30:00"],
      "oldkeys": {
        "keynames": ["id"],
        "keyvalues": [1]
      },
      "oldvalues": [1, "John Doe", "john@example.com", "2024-01-15T10:30:00"]
    }
  ]
}

Trade-offs:

✅ Easy to parse in any language
✅ Human-readable format
✅ Supported on most managed services (RDS, Cloud SQL)
❌ Higher overhead than binary formats
❌ Can struggle with very large transactions

decoderbufs

decoderbufs uses Protocol Buffers for efficient binary serialization.

What our user name update looks like:

# Binary protobuf message (conceptual)
RowMessage {
  transaction_id: 12345
  table: "public.users"
  op: UPDATE
  new_tuple: {
    columns: [
      {name: "id", type: INTEGER, value: 1},
      {name: "name", type: TEXT, value: "John Smith"},
      {name: "email", type: TEXT, value: "john@example.com"}
    ]
  }
  old_tuple: {
    columns: [
      {name: "name", type: TEXT, value: "John Doe"}
    ]
  }
}

Trade-offs:

✅ Very efficient binary format
✅ Schema-defined structure
✅ Great for high-throughput scenarios
❌ Requires building and installing
❌ Not available on most managed services
❌ More complex to work with

Specialized plugins

wal2mongo: Specifically designed for replicating to MongoDB, outputs MongoDB-compatible JSON operations.

decoder_raw: Outputs changes as raw SQL statements that you could execute on another database.

Choosing the right plugin

Your choice often comes down to a few practical factors:

Environment constraints

Managed services (AWS RDS, Google Cloud SQL, Azure): You're typically limited to pgoutput, test_decoding, and wal2json. Choose pgoutput for Postgres-to-Postgres replication or wal2json for external integrations.

Self-hosted: You have full flexibility. Consider decoderbufs for high-performance scenarios or stick with pgoutput for simplicity.

Performance requirements

For high-volume scenarios:

pgoutput - Best overall choice for most use cases
decoderbufs - If you need non-Postgres output and can manage the complexity
wal2json - Convenient but can be a bottleneck at scale

Output plugin vs. final format

One common confusion: the output plugin format isn't necessarily the format your application sees.

For example, with Sequin:

Debezium can use pgoutput to receive binary data from Postgres
Then convert it to JSON for delivery to sinks
Your application sees the JSON format, not the pgoutput format

Standing up Debezium Server for Postgres CDC

Eric Goldman — Wed, 14 May 2025 18:53:54 +0000

In this tutorial, you'll set up a complete CDC pipeline using Debezium Server (version 3.1) to capture changes from a PostgreSQL database and stream them directly to a webhook endpoint. Unlike the Kafka Connect runtime, Debezium Server provides a lightweight standalone application that can send change events directly to various destinations without requiring a Kafka cluster. Keep in mind that your trading out the overhead of Kafka for less scale, performance, and delivery guarantees.

What you'll learn:

How to configure PostgreSQL for logical replication
How to set up Debezium Server to capture database changes
How to stream these changes directly to a webhook endpoint
How to observe and work with CDC events from your database

The Architecture

Here's what you're building:

A PostgreSQL database with logical replication enabled
Debezium Server running as a standalone application
A webhook endpoint (using webhook.site) that receives the change events
A simple "customers" table that you'll monitor for changes

When you're done, any change to the customers table will be captured by Debezium Server and sent as a JSON event to your webhook endpoint in real-time. This setup provides a foundation for building event-driven architectures without the complexity of managing a Kafka cluster.

Step 1: Configure PostgreSQL for logical replication

Ensure logical replication is enabled on your Postgres database:

psql -U postgres -c "SHOW wal_level;"

You should see logical in the output. If not, run the following commands to enable logical replication:

# Connect to PostgreSQL as the postgres user to modify system settings
sudo -u postgres psql -c "ALTER SYSTEM SET wal_level = logical;"
sudo -u postgres psql -c "ALTER SYSTEM SET max_replication_slots = 10;"
sudo -u postgres psql -c "ALTER SYSTEM SET max_wal_senders = 10;"

# Restart PostgreSQL to apply changes
# For Linux (systemd):
sudo systemctl restart postgresql

# For macOS (Homebrew):
brew services restart postgresql

These commands:

Set the Write-Ahead Log (WAL) level to "logical", enabling detailed change tracking
Configure replication slots to allow Debezium to track its position
Increase the number of WAL sender processes that can run simultaneously

Step 2: Create a database user and sample data

Debezium requires a PostgreSQL user with replication privileges and a table to monitor.

# Create a dedicated user for Debezium
psql -U postgres -c "CREATE ROLE dbz WITH LOGIN PASSWORD 'dbz' REPLICATION;"

# Create a sample database
createdb -U postgres inventory

# Create a sample table
psql -d inventory -U postgres <<'SQL'
CREATE TABLE customers (
  id SERIAL PRIMARY KEY,
  name TEXT NOT NULL,
  email TEXT UNIQUE,
  created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);

-- Set REPLICA IDENTITY to FULL to capture old values on updates and deletes
ALTER TABLE customers REPLICA IDENTITY FULL;

-- Grant necessary permissions to the dbz user
GRANT ALL PRIVILEGES ON DATABASE inventory TO dbz;
GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA public TO dbz;
GRANT ALL PRIVILEGES ON ALL SEQUENCES IN SCHEMA public TO dbz;
SQL

The REPLICA IDENTITY FULL setting ensures that both old and new values are captured for updates and deletes, which is crucial for comprehensive change tracking.

Step 3: Download and set up Debezium Server

Debezium Server is a standalone application that connects to PostgreSQL and forwards change events to various sinks. Let's download and extract it:

# Download Debezium Server
curl -L -o debezium-server.zip https://repo1.maven.org/maven2/io/debezium/debezium-server-dist/3.1.1.Final/debezium-server-dist-3.1.1.Final.zip

# Extract the archive
unzip debezium-server.zip

# Navigate to the Debezium Server directory
cd debezium-server

You'll now have a directory structure containing:

run.sh - The script to start Debezium Server
lib/ - JAR files for Debezium and its dependencies
config/ - Configuration directory

Step 4: Create a webhook endpoint

To make it easy to standup and test Debezium Server, use webhook.site to quickly test webhook delivery:

Open https://webhook.site in your browser
A unique URL will be automatically generated for you (it looks like https://webhook.site/XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX)
Copy this URL - you’ll use it in the Debezium configuration
Keep this browser tab open to see events as they arrive

Step 5: Configure Debezium Server

Create or modify the application.properties file in the config/ directory to tell Debezium where to connect and where to send events:

# Create or modify the configuration file
# PostgreSQL source connector configuration
debezium.source.connector.class=io.debezium.connector.postgresql.PostgresConnector
debezium.source.offset.storage.file.filename=data/offsets.dat
debezium.source.offset.flush.interval.ms=0
debezium.source.database.hostname=localhost
debezium.source.database.port=5432
debezium.source.database.user=dbz
debezium.source.database.password=dbz
debezium.source.database.dbname=inventory
debezium.source.database.server.name=inventory-server
debezium.source.schema.include.list=public
debezium.source.table.include.list=public.customers

# Set a unique replication slot name to avoid conflicts
debezium.source.slot.name=debezium_tutorial

# Topic prefix configuration
debezium.source.topic.prefix=inventory-server

# For capturing all changes including updates and deletes in full
debezium.source.tombstones.on.delete=false

# Initial snapshot configuration
debezium.source.snapshot.mode=initial

# HTTP sink (webhook) configuration
debezium.sink.type=http
debezium.sink.http.url=YOUR_WEBHOOK.SITE_URL_HERE
debezium.sink.http.timeout.ms=10000

# JSON formatter
debezium.format.value=json
debezium.format.key=json

Replace YOUR_WEBHOOK_URL_HERE with the URL you copied from webhook.site.

Step 6: Start Debezium Server

Now that everything is configured, start Debezium Server:

# Make the run script executable
chmod +x run.sh

# Start Debezium Server
./run.sh

Prepare for a wall of Java logs. You should see output indicating that Debezium Server is starting, with messages about Quarkus, the connector, and eventually reaching a "started" state.

Common startup messages include:

Quarkus initialization
PostgreSQL connector configuration
Connection to the database
Snapshot process (if this is the first run)
HTTP sink initialization

Step 7: Test the setup with database changes

Open a new terminal (keep Debezium Server running in the first one) and execute some changes to the customers table:

# Insert some test data
psql -d inventory -U postgres <<'SQL'
INSERT INTO customers (name, email) VALUES 
  ('Alice Johnson', 'alice@example.com'),
  ('Bob Smith', 'bob@example.com');
SQL

# Update a record
psql -d inventory -U postgres -c "UPDATE customers SET email = 'alice.new@example.com' WHERE name = 'Alice Johnson';"

# Delete a record
psql -d inventory -U postgres -c "DELETE FROM customers WHERE name = 'Bob Smith';"

Step 8: Observe the results

Switch back to your webhook.site browser tab. You should see several POST requests that correspond to the database operations you just performed:

Two insert events (one for each new customer)
An update event (when you changed Alice's email)
A delete event (when you removed Bob's record)

Each event contains a JSON payload with details about the operation and the data. For example, an insert event might look like this:

{
  "schema": { /* schema information */ },
  "payload": {
    "before": null,
    "after": {
      "id": 1,
      "name": "Alice Johnson",
      "email": "alice@example.com",
      "created_at": "2023-05-06T10:15:30.123456Z"
    },
    "source": {
      /* metadata about the event source */
      "db": "inventory",
      "table": "public.customers",
      "operation": "c",
      "ts_ms": 1683367230123
    },
    "op": "c",
    "ts_ms": 1683367230456
  }
}

The op field indicates the operation type:

c for create (insert)
u for update
d for delete
r for read (during initial snapshot)

For updates, both before and after fields are populated, showing the previous and new values.

How it works

Let's understand what's happening under the hood:

PostgreSQL logical replication: The WAL settings enable PostgreSQL to maintain a log of changes that can be read by external processes.
Debezium Server: Acts as a standalone change data capture service that:
- Connects to PostgreSQL using the configured credentials
- Reads the WAL stream to detect changes
- Converts database changes to structured JSON events
- Forwards these events to the configured sink (webhook in our case)
Webhook endpoint: Receives HTTP POST requests containing the change events as JSON payloads.

Next steps and variations

Now that you have a working Debezium setup, here are some ways to expand and customize it:

Monitor multiple tables

To track changes from additional tables, adjust the debezium.source.table.include.list property in application.properties:

debezium.source.table.include.list=public.customers,public.orders,public.products

Transform events before sending

Debezium supports Single Message Transforms (SMTs) to modify events before they're sent. For example, to rename a field:

# Add this to application.properties
debezium.transforms=rename
debezium.transforms.rename.type=org.apache.kafka.connect.transforms.ReplaceField$Value
debezium.transforms.rename.renames=email:contact_email

Conclusion

You've successfully set up Debezium Server to capture and stream PostgreSQL changes to a webhook endpoint. This foundation can be extended to build robust event-driven architectures, real-time data pipelines, and more.

Want to skip all this complexity? Check out Sequin for hassle-free change data capture and event streaming without the maintenance burden.

Streaming Postgres Changes with Debezium and Kafka Connect: A Hands-On Tutorial

Eric Goldman — Tue, 06 May 2025 01:32:01 +0000

Looking for a simpler solution? If you're reading this tutorial, you want to reliably stream changes out of your Postgres database. You’ve turned to Debezium, only to discover that the official Debezium documentation leaves more to be desired. I hit the same snag, and decided to save future developers similar toil by writting this guide. If you are looking for an easier, faster alternative to Debezium, we built Sequin to provide the same guarantees of Debezium while eliminating all this complexity.

Introduction

In this tutorial, you'll set up a complete CDC pipeline using Debezium (version 3.1) with Kafka Connect to capture changes from a PostgreSQL database. By the end, you'll have a working system that streams every insert, update, and delete operation from your database into Apache Kafka topics.

What you'll learn:

How to set up a local development environment for Debezium with Docker Compose
How to configure PostgreSQL for logical replication
How to set up a Debezium connector to capture changes
How to observe and work with CDC events from your database

Let's get started!

Prerequisites

Before diving in, make sure you have:

Docker Engine and Docker Compose v2 installed on your system
curl command-line tool for making HTTP requests
Basic familiarity with PostgreSQL and command-line operations

You'll be doing everything in docker and your CLI - Debezium doesn't come with a UI or additional tooling.

The Architecture

Here's what you're setting up:

A PostgreSQL database with logical replication enabled
Apache Kafka (which requires ZooKeeper in this version) for message streaming
Kafka Connect with the Debezium PostgreSQL connector to provide CDC on the Postgres database and stream those changes to the aformentioned Kafka topic
A simple "customers" table that you'll monitor for changes

When you're done, any change to the customers table will be captured by Debezium and sent as an event to a Kafka topic, which you can then consume from any application.

Step 1: Setting Up the Infrastructure with Docker Compose

First, create your environment using Docker Compose. This will spin up all the necessary services in isolated containers.

Create a docker-compose.yml file

Create a new directory and then create a new file called docker-compose.yml:

mkdir debezium-example
cd debezium-example
touch docker-compose.yml

Copy and paste the following into the docker-compose.yml:

services:
  zookeeper:
    image: quay.io/debezium/zookeeper:3.1
    ports: ["2181:2181"]

  kafka:
    image: quay.io/debezium/kafka:3.1
    depends_on: [zookeeper]
    ports: ["29092:29092"]
    environment:
      ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENERS: INTERNAL://0.0.0.0:9092,EXTERNAL://0.0.0.0:29092
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,EXTERNAL://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  connect:
    image: quay.io/debezium/connect:3.1
    depends_on: [kafka]
    ports: ["8083:8083"]
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: connect_configs
      OFFSET_STORAGE_TOPIC: connect_offsets
      STATUS_STORAGE_TOPIC: connect_statuses
      KEY_CONVERTER_SCHEMAS_ENABLE: "false"
      VALUE_CONVERTER_SCHEMAS_ENABLE: "false"

  postgres:
    image: debezium/postgres:15
    ports: ["5432:5432"]
    command: postgres -c wal_level=logical -c max_wal_senders=10 -c max_replication_slots=10
    environment:
      POSTGRES_USER: dbz
      POSTGRES_PASSWORD: dbz
      POSTGRES_DB: example

Understanding the docker-compose.yml

Briefly, this file defines the following services - all of which are needed to run Debezium:

ZooKeeper: Provides distributed configuration and synchronization for Kafka
Kafka: The message broker that will store your change data events.
Kafka Connect: The framework that runs your Debezium connector
PostgreSQL: Your database with logical replication enabled

Notice the PostgreSQL configuration:

You're using the debezium/postgres:15 image, which comes with the necessary logical decoding plugins
You set wal_level=logical to enable logical replication
You configure max_wal_senders and max_replication_slots to allow multiple replication connections

Start the containers

Back in your terminal, ensure you are in the debezium-example directory containing your docker-compose.yml file, and run:

docker compose up -d

This command starts all the services in detached mode. To verify that all containers are running:

docker compose ps

You should see all four containers (zookeeper, kafka, connect, and postgres) with a status of "Up".

$ docker ps
CONTAINER ID   IMAGE                            COMMAND                  CREATED          STATUS          PORTS                                        NAMES
c218e8a9fc67   quay.io/debezium/connect:3.1     "/docker-entrypoint.…"   59 minutes ago   Up 59 minutes   8778/tcp, 0.0.0.0:8083->8083/tcp, 9092/tcp   debezium-connect-1
5ffe2fc31745   quay.io/debezium/kafka:3.1       "/docker-entrypoint.…"   59 minutes ago   Up 59 minutes   9092/tcp, 0.0.0.0:29092->29092/tcp           debezium-kafka-1
1b1de4194458   debezium/postgres:15             "docker-entrypoint.s…"   59 minutes ago   Up 59 minutes   0.0.0.0:5432->5432/tcp                       debezium-postgres-1
d5d87a35b013   quay.io/debezium/zookeeper:3.1   "/docker-entrypoint.…"   59 minutes ago   Up 59 minutes   2888/tcp, 0.0.0.0:2181->2181/tcp, 3888/tcp   debezium-zookeeper-1

Step 2: Preparing PostgreSQL for CDC

Now that the required infrastructure is running, you need to configure PostgreSQL for change data capture. You’ll connect to the Postgres instance you just created and add a new replication user for Debezium, set up a demo table, and configure the table for full row captures.

Create a replication user

Create the demo table

Create a simple customers table that we'll monitor for changes:

docker compose exec postgres \
  psql -U dbz -d example \
  -c "CREATE TABLE customers (id SERIAL PRIMARY KEY, name VARCHAR(255), email VARCHAR(255));"

This table has three columns:

id: An auto-incrementing primary key
name: A customer's name
email: A customer's email address

If you'd like to explore the database more directly in a SQL client like TablePlus, you can connect to it using any PostgreSQL client with these connection details:

Host: localhost

Port: 5432

Database: example

User: dbz

Password: dbz

SSL Mode: disable

Or using a connection string:
postgresql://dbz:dbz@localhost:5432/example?sslmode=disable
This allows you to explore the database schema, run queries, and make changes that will be captured by Debezium.

Configure full row images

By default, PostgreSQL's logical replication only includes the primary key and changed columns in update events. To get the full "before" and "after" state of rows, you need to set the REPLICA IDENTITY to FULL:

docker compose exec postgres \
  psql -U dbz -d example \
  -c "ALTER TABLE customers REPLICA IDENTITY FULL;"

This setting ensures that when a row is updated or deleted, the entire row's data (before the change) is included in the WAL (Write-Ahead Log) entry, giving you complete information about the change in the resulting Kafka message produced by Debezium.

Step 3: Setting Up the Debezium Connector

Now, configure and register the Debezium PostgreSQL connector with Kafka Connect.

Create the connector configuration

Create a file named register-postgres.json with the following content:

{
  "name": "example-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "dbz",
    "database.password": "dbz",
    "database.dbname": "example",
    "topic.prefix": "example",
    "slot.name": "example_slot",
    "publication.autocreate.mode": "filtered",
    "table.include.list": "public.customers"
  }
}

This configuration:

Names the connector example-connector
Specifies the PostgreSQL connection details
Sets a topic prefix of example (resulting in a topic named example.public.customers)
Creates a replication slot named example_slot
Configures the publication to only include the tables you specify
Limits change capture to only the public.customers table

Register the connector

curl -X POST -H "Content-Type: application/json" \
     --data @register-postgres.json \
     http://localhost:8083/connectors

If successful, you should receive a JSON object that looks similar to the configuration in your register-postgres.json file. This means that Kafka Connect has started the Debezium connector, which is now listening for changes to our customers table.

Verify the connector is running

You can check the status of your connector:

curl -s http://localhost:8083/connectors/example-connector/status | jq

{
  "name": "example-connector",
  "connector": {
    "state": "RUNNING",
    "worker_id": "172.20.0.5:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "RUNNING",
      "worker_id": "172.20.0.5:8083"
    }
  ],
  "type": "source"
}

You should see that the connector is in the RUNNING state. If you encounter any issues, you can check the Kafka Connect logs:

docker compose logs -f connect

Step 4: Generating and Observing Change Events

Now for the exciting part! Make some changes to your database and watch the CDC events flow into Kafka.

Start a Kafka consumer

First, open a terminal window and start a console consumer to watch the Kafka topic:

docker compose exec kafka /kafka/bin/kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic example.public.customers --from-beginning

This command connects to Kafka and subscribes to the example.public.customers topic, displaying all messages as they arrive.

Insert a new customer

In a new terminal window, insert a record into our customers table:

docker compose exec postgres \
  psql -U dbz -d example \
  -c "INSERT INTO customers(name,email) VALUES ('Alice','alice@example.com');"

In your consumer terminal, you should see a JSON message appear that looks something like this:

{
  "before": null,
  "after": {
    "id": 1,
    "name": "Alice",
    "email": "alice@example.com"
  },
  "source": {
    "version": "3.1.0.Final",
    "connector": "postgresql",
    "name": "example",
    "ts_ms": 1620000000000,
    "snapshot": "false",
    "db": "example",
    "sequence": "[\"12345678\",\"12345678\"]",
    "schema": "public",
    "table": "customers",
    "txId": 123,
    "lsn": 12345678
  },
  "op": "c",
  "ts_ms": 1620000000001
}

This is the Debezium change event format, also known as the "envelope." It contains:

before: The previous state of the row (null for inserts)
after: The new state of the row
source: Metadata about the event source
op: The operation type ("c" for create/insert)
ts_ms: Timestamp of when the event was processed

Update the customer

Now let's update the customer record:

docker compose exec postgres \
  psql -U dbz -d example \
  -c "UPDATE customers SET name='Alice Updated' WHERE id=1;"

In your consumer terminal, you should see another JSON message, this time with both before and after fields populated:

{
  "before": {
    "id": 1,
    "name": "Alice",
    "email": "alice@example.com"
  },
  "after": {
    "id": 1,
    "name": "Alice Updated",
    "email": "alice@example.com"
  },
  "source": { /* ... */ },
  "op": "u",
  "ts_ms": 1620000000002
}

Note that op now has a value of "u" for update, and both before and after states are included.

Delete the customer

Finally, let's delete the customer:

docker compose exec postgres \
  psql -U dbz -d example \
  -c "DELETE FROM customers WHERE id=1;"

You'll see a final message with op set to "d" (delete):

{
  "before": {
    "id": 1,
    "name": "Alice Updated",
    "email": "alice@example.com"
  },
  "after": null,
  "source": { /* ... */ },
  "op": "d",
  "ts_ms": 1620000000003
}

Notice that for deletes, after is null since the row no longer exists after the operation.

Step 6: Cleaning Up

When you're finished experimenting, clean up the environment:

docker compose down -v

This stops all containers and removes volumes created by Docker Compose.

Understanding the Architecture

Now that you've seen Debezium in action, it’s helpful to discuss what is happening behind the scenes.

PostgreSQL Write-Ahead Log (WAL): When data changes in PostgreSQL, those changes are written to the WAL for durability. With logical replication enabled, these changes can be decoded into logical change events (we go into more detail in this post).
Debezium PostgreSQL Connector: Debezium creates a replication slot in PostgreSQL and subscribes to changes. It reads from the WAL, transforms the binary changes into structured events, and sends them to Kafka.
Kafka Connect: This framework manages the connector lifecycle and ensures reliable delivery of events to Kafka, handling failures and offsets.
Kafka Topics: Each table's changes are published to a dedicated topic, allowing consumers to subscribe only to the tables they care about.

Conclusion

You've successfully set up a change data capture pipeline using Debezium and Kafka Connect. You've seen how to capture inserts, updates, and deletes from PostgreSQL and stream them to Kafka in real-time.

In this Kafka Connect runtime, Debezium relies on Kafka (and the Kafka Exosystem) to provide capabilities like single message transforms, message retention, and efficient scaling. Now that the system is running, get familiar with these tool to get set up.

From here, you'll want to tailor this template to your Postgres database. Ensure you enable logical replication in your Postgres instance, create a user for Debezium, and then buckle up for Java logs.

A developer's reference to Postgres change data capture (CDC)

Eric Goldman — Thu, 17 Apr 2025 02:32:40 +0000

As a developer researching change data capture (CDC) on Postgres, what do you need to know?

We’ve worked with hundreds of developers as they implement CDC specifically on Postgres. We’ve written extensively on the topic. This post aims to synthesize our learnings to help you build a better understanding of CDC, weigh different implementation options, and find the best CDC implementation for your use case.

What is Change Data Capture (CDC)

As the name suggests, change data capture is a process that detects every incremental change to a database, and then delivers (i.e. captures) those changes in a downstream system. When done correctly, the other system can process these changes to perfectly replicate the original data set.

In practice, this means that every insert, update, and delete in the Postgres database needs to be captured and delivered in order. No alteration can be missed. The physics of never missing a change and guaranteeing delivery makes building a reliable CDC system challenging.

Luckily, PostgreSQL 10+ comes with logical replication built in. The database records every change in the Write Ahead Log (WAL). You can tap into the WAL and decode the messages to capture every insert, update, and delete. Moreover, Postgres writes messages to the WAL during normal operation, so processing changes through the WAL adds no additional overhead to the database.

While the WAL is the most reliable way to interface with Postgres to power CDC, it isn’t the only option nor is it necessarily the easiest to work with.

When to reach for CDC

Almost every data intensive application includes a CDC component. (In fact, we’ve found that almost every company with 10 or more engineers has either built or bought a CDC solution.)

As systems grow, teams begin to move work away from a single Postgres instance towards specialized tools. In this effort, common use cases begin to emerge that require CDC:

Database replication: Reliably mirror data across databases with greater flexibility than traditional Postgres read-replicas. CDC enables the primary database to operate independently of replicas while providing powerful tools for transforming data during transmission.
Real-time analytics: Stream changes directly to OLAP databases and data warehouses, enabling immediate insights and analytics without impacting production database performance.
Cache management: Ensure application caches and search indexes remain perfectly synchronized with source data, eliminating staleness issues while reducing unnecessary refresh operations.
Microservice consistency: Preserve data integrity across distributed systems by propagating changes to specialized services while maintaining strong transactional guarantees.
Event-driven automation: Trigger workflows and jobs with immediate, transaction-guaranteed execution, enabling responsive systems that act on data changes as they occur.
Audit logging: Capture every database change for compliance requirements or to power user-facing features like detailed history views and precision rollbacks.

Beyond use cases, CDC also helps reduce technical debt while setting a sturdy foundation for scale.

For instance, imagine that one system is constantly polling a table for changes while another uses triggers to log changes from the same table. The load on the database is unnecessarily high, and the process of testing and building new features becomes unnecessarily complex. CDC provides a consistent pattern to detect changes on the same table independently - without any additional burden on the database.

Postgres CDC performance considerations

A good change data capture system will provide a set of guarantees around reliability, performance, and data integrity. The right CDC system will completely abstract a set of edge cases and bugs so your team can build with confidence.

Here are some important characteristics to consider as you build or buy a CDC solution:

Delivery guarantees: Can you expect every change in the database to be delivered. In many CDC implementations, it’s possible for a message to be dropped (at-most-once delivery) or delivered more than once (at-least-once delivery). Ideally, the CDC system will provide an “exactly-once processing guarantee” by providing sophisticated deduplication and a method for acknowledging when a change has been delivered and processed. For Postgres CDC use cases where consistency is critical (i.e. replication, cache invalidation, etc), then an at-least-once delivery guarantees are a bare minimum, but exactly-once processing is ideal.

Throughput: What volume of change events can the CDC system handle within a given timeframe. This is best measured as bandwidth (i.e. MB/sec), but operations per second can be a simple proxy. Importantly, your CDC solution should be able to keep pace with the peak IOPS of your Postgres database, and ideally with room to spare. If your CDC solution is too slow, it’s not uncommon for CDC solutions to fall behind during peak load, and then never recover.

Latency: The time delay between a change being made in your database and then appearing in your target system. For some use cases (like analytics or reporting) a latency of hours or even days is sufficient. For other customer-facing use cases (i.e. updating a search index) millisecond latency ensures that results are accurate. It’s best to measure the p95 or p99 latency of your CDC service to understand the tolerances you can tune your system to. It’s also important to evaluate latency in the context of throughput to understand how your CDC system can fall behind, which is the ultimate impact to latency.

Retries and error handling: How does your system handle transient errors. One poison pill message can immediately jam and crash your CDC pipeline - which in some circumstances (i.e. replication) can be ideal. Other implementations will queue the errant message into a dead letter queue (DLQ), log an alert, and attempt to redeliver with exponential retries.

Ordering: Maintaining the correct order of change events is critical to avoid data corruption or inconsistencies. This is particularly true when there are interdependent operations that can affect the state of record. Enforcing strict ordering often comes at the expense of throughput and latency as enforcing ordering often makes parallel processing harder. Great CDC options allow you to configure what kinds of changes need to be processed in order to strike the right balance between speed and structure.

Schema evolution: The ability to handle modifications to the PostgreSQL database schema, such as adding, altering, or removing columns. These kinds of changes can break downstream services and either need to be handled gracefully (i.e. propagate the change down stream) or should proactively halt the system.

Snapshots: How does the CDC service capture the initial state of the Postgres table. Snapshotting (a.k.a backfilling) isn’t just about initializing the CDC process - it’s often the first thing you’ll need to do after an incident. The ability to both completely snapshot a table or target just a subset of the table (using SQL) is very helpful. A great CDC system needs to be able to snapshot a massive table while simultaneously capturing new changes.

Database load: How much CPU and storage capacity does the CDC solution consume on the database. Systems that poll or use triggers to detect changes can slow the database down significantly. Others that read through the WAL add virtually no overhead to the database in normal operation, but if they disconnect from the replication slot can rapidly consume available storage.

Monitoring and observability: Change data capture often becomes a critical component of your application - so you’ll want to be able to measure and observe each of the characteristics mentioned above. Great CDC solutions can plug into your existing monitoring and observability tooling. The difference between cryptic, convoluted logs and clear traceable errors can make all the difference.

Before building or buying a CDC solution, consider what performance characteristics are required for your use case. Often, picking a great CDC tool from the start can quickly set you up with a simpler architecture, higher performance, and an easier build.

Build your own Postgres CDC

For developers looking to implement Change Data Capture with PostgreSQL directly, several methods are available, each with its own set of advantages and disadvantages. These approaches offer varying levels of control and complexity, as detailed in our guide All the ways to capture changes in Postgres.

One common method involves using triggers with an audit table. This approach entails creating database triggers that activate upon data modification events (INSERT, UPDATE, DELETE) on the tables of interest. When a change occurs, the trigger fires and typically records the details of the change in a separate audit log table. Triggers are reliable as they operate within the SQL system and offer real-time capture with a high degree of customization for various event types. However, they can introduce performance overhead on the primary database and require careful management of the audit log table.

Another straightforward method is polling for changes. This involves adding a dedicated timestamp column to the tables and periodically querying for records that have been modified since the last check, based on the timestamp. This approach is simple to implement for basic scenarios and doesn't require complex database configurations. However, it necessitates the presence of a timestamp column, can be resource-intensive with frequent polling, and cannot reliably capture DELETE events unless soft deletions are employed.

Depending on your use case, you might be able to get away with Listen/Notify. This built-in PostgreSQL feature implements a publish-subscribe pattern where database sessions can listen on channels and receive real-time notifications of data changes. By setting up triggers that fire on INSERT, UPDATE, or DELETE operations, you can broadcast JSON payloads containing change details to listening applications. While simple to implement and offering immediate notification capabilities, Listen/Notify has key limitations: it provides only "at-most-once" delivery semantics (requiring active connections when notifications occur), limits payloads to 8000 bytes, and offers no persistence for missed messages. These constraints make it suitable primarily for lightweight change detection scenarios or as a complement to more robust CDC methods, rather than for mission-critical systems requiring guaranteed delivery.

Finally, logical replication (via the WAL) stands out as a robust and efficient method. It's a built-in feature of PostgreSQL that allows for the selective replication of data changes at the table level. Leveraging log-based CDC, it provides real-time, event-driven capture of all change types with minimal impact on the database and broad support across different PostgreSQL environments. Logical replication has become the favored approach for PostgreSQL CDC due to its optimal balance of efficiency, reliability, and comprehensive change capture capabilities.

While each of these approaches captures changes in the database, that is just the first step in building a CDC system. You then need to deliver those changes to other systems with sufficient performance guarantees (see above). This often adds significant complexity and work, which is why most teams choose an off the shelf approach.

Off the shelf Postgres CDC

To implement PostgreSQL CDC without the complexities of building a solution from scratch, a variety of robust tools and platforms are available, each offering unique features and capabilities. A detailed overview of these options can be found at https://blog.sequinstream.com/choosing_the_right_real_time_postgres_cdc_platform/.

Open source

Among the open-source tools, Sequin and Debezium are the go to options.

Sequin is a fast, simple, modern open source CDC platform tuned specifically to Postgres. It is known as the fastest, highest throughput change data capture solution for Postgres. It supports many different destinations with exactly-once processing guarantees. It’s easy to self-host via Docker and provides prometheus endpoints for monitoring and observation. It comes with an easy to use web console, CLI, and API as well as helpful developer tools. If you don’t want to self-host, you can also use Sequin Cloud.

Debezium is a widely adopted distributed platform specifically designed for CDC. Built on top of Apache Kafka and Kafka Connect, Debezium captures changes in real-time using logical replication and streams them to Kafka topics, ensuring exactly-once processing. However, it requires the setup and management of Kafka, which can introduce operational complexity. It’s notoriously difficult to configure, maintain, and debug - and its future is unclear since it’s been moved from Red Hat support to the Commonhause Foundation.

Hosted

For teams seeking the benefits of CDC without the operational overhead, several hosted solutions provide managed CDC capabilities.

Sequin Cloud is a fully hosted, highly available deployment of Sequin. It’s immediately configured to scale and runs in regions all around the world. It’s priced to be the most economical CDC solution for Postgres and comes with additional features around tracing, alerting, team management, and disaster recovery. With Sequin Cloud, you pay for just the data you process through Sequin - completely abstracting the underlying infrastructure.

Decodable offers a hosted Debezium plus Apache Flink stack, making it ideal for organizations already invested in these technologies but wanting them managed. It captures every database change with strong deliverability guarantees and includes monitoring and schema management tools. While easier than self-hosting, using Decodable still requires learning how to configure pipelines and apply SQL transforms in their dashboard, with Flink pipelines presenting a moderate learning curve. Cost-wise, Decodable is moderately expensive, with one pipeline potentially comprising several "tasks" that can each cost hundreds of dollars monthly.

Confluent provides two options: Confluent Debezium and Direct JDBC Postgres connector. The Debezium version is a fully hosted implementation with enterprise security, compliance, monitoring, and scaling tools. While you won't need to manage the deployment directly, you'll still handle the complex configuration of Debezium and Kafka topics. The Direct JDBC connector uses polling rather than logical replication, meaning it only captures inserts and updates (not deletes) with some delay. While easier to set up, it's less powerful. Both options come with enterprise pricing, trading engineering time for an expensive infrastructure product.

Estuary is a fully managed CDC solution that simplifies real-time data pipelines. It provides a wide selection of pre-built connectors that allow you to quickly connect your sources and destinations. Notably, Estuary loads data from into an intermediary data source so that subsequent backfills are fast with minimal overhead on the database.

Streamkap is a serverless streaming platform designed for real-time CDC. It leverages Apache Kafka and Debezium with connectors to enable fast and efficient data ingestion. Streamkap specializes in delivering CDC data to various destinations, including ClickHouse, Snowflake, and others.

Striim positions itself as an enterprise solution with high reliability but at premium cost. It provides CDC with transformations, schema management, delivery guarantees, and security features designed for Fortune 1,000 companies. Using Striim requires learning their proprietary TQL (Transform Query Language) for data transformations and StreamApps framework for pipeline configuration. While well-documented, these tools have a steep learning curve and are unique to Striim's ecosystem. Budget-wise, Striim requires an all-in enterprise contract.

ETL Providers

Traditional ETL providers like Fivetran and Airbyte offer CDC capabilities, but it's important to note that they typically provide batch CDC rather than real-time solutions. Changes may take minutes to appear in your stream or queue, and they may not maintain atomic change records.

These solutions are primarily intended for non-operational, analytics use cases rather than real-time operational requirements. They offer scheduled, batch delivery of database changes to various destinations, including streams like Kafka. While they're relatively easy to set up, they offer limited configuration options when used for CDC.

Cost can be a significant factor with these providers, as pricing is typically based on row count, which can quickly become expensive for high-volume databases. If your use case can tolerate batch processing and you're primarily focused on analytics, these tools provide a simple solution with minimal technical expertise required.

Cloud providers

The major cloud infrastructure providers offer CDC products that work within their ecosystems. AWS DMS (Database Migration Service), GCP Datastream, and Azure Data Factory can all be configured to stream changes from Postgres to other infrastructure within their respective platforms.

These solutions can be effective if your organization is already committed to a specific cloud provider and comfortable with their tooling. They support logical replication with real-time capture of inserts, updates, and deletes, though delivery guarantees can vary based on configuration and provider.

Setting up CDC through your cloud provider typically requires navigating their web console, permissions systems, tooling, and logging to establish pipelines. You'll need familiarity with the provider's settings and configurations. From a cost perspective, these solutions typically charge for compute hours and data transfer, which can be difficult to predict and may accumulate quickly. Additionally, these setups can become brittle over time as settings and dependencies are distributed across different services.

While convenient for teams already invested in a particular cloud ecosystem, these solutions may not offer the same level of specialized functionality as dedicated CDC platforms.

Conclusion

Change Data Capture has become an essential component of modern data architectures, enabling real-time integration across increasingly specialized and distributed systems. It’s evolved from a bespoke approach to data integration, to a more specialized piece of infrastructure with defined performance characteristics and features. This post helps you map your use case and requirements to CDC implementation options and tools. Reach out if you have more questions about Postgres CDC!

How Debezium captures changes from Postgres

Eric Goldman — Thu, 17 Apr 2025 02:25:03 +0000

Debezium is a popular solution for streaming changes from Postgres. It was amongst the first generation of tools purpose built to help developers implement a change data capture (CDC) pattern. While Debezium introduces complex dependencies like Kafka to operate, in total it abstracts away many of the challenges around detecting and delivering every change in your database.

This post aims to help developers understand how Debezium works with Postgres so you can better architect your system.

What is Debezium

Debezium captures row-level changes in databases and streams them to Kafka topics. It's implemented as a set of Kafka Connect source connectors, with each connector ingesting changes from a different database system using that database's native logging capabilities.

Unlike polling or dual-write approaches, Debezium's log-based CDC implementation:

Guarantees that all data changes are captured without gaps. Including every insert, update, and delete.
Produces change events with minimal delay
Requires no modifications to your data models (no "Last Updated" columns)
Preserves original record states and transaction metadata

Debezium then transforms each row-level operation into change event records, which are then streamed to distinct Kafka topics. Applications can consume these event streams to react to data changes in near real-time, enabling a wide range of use cases from cache invalidation to analytics pipelines.

How Does Postgres Emit Logs to Debezium

When integrating with Postgres, Debezium interfaces with Postgres’ write-ahead log (WAL). The WAL is essentially a sequential log of all database modifications. Importantly, any change in the Postgres database is first added to the log and flushed to persistent storage (hence write-ahead) before the change is actually committed.

Consider what happens when you make an INSERT in Postgres:

Open transaction: Postgres validates the insert statement, permissions, and creates an execution plan. Ultimately, a transaction is opened for the insert.
Create WAL record: Before modifying any data in memory, Postgres creates a WAL record that contains the details of the operation. The WAL record includes a Log Sequence Number (LSN) that uniquely identifies the position / order of the operation.
In-memory buffer: The WAL record is written to an in-memory WAL buffer just prior to the actual row being inserted into the in-memory buffer for the table.
Flush to disk: When the in-memory transaction commits, the WAL buffer is flushed to disk and the transaction is marked as committed. At this point, your client receives a confirmation that the insert completed. This is also when a client connected to the WAL (i.e. Debezium) receives the WAL entry for the insert.
Checkpointing: At some later point the change made in memory is also written to the data files.

Debezium connects to Postgres via a replication slot to read the records in the WAL (at step four above). The WAL records in the slot are decoded from binary using the pgoutput plugin. This approach to detecting changes comes with important benefits:

Exact order: By using the LSN as a cursor, Debezium is able to process changes in order.
Only completed transactions: Changes are only flushed to the WAL when the transaction commits, so Debezium never captures partial or rolled-back transactions.
Minimal database overhead: Changes are committed to the WAL as a normal part of the transaction. As long as Debezium pulls and flushes messages from the replication slot (foreshadowing!), no new overhead is added to the database.

The Debezium Capture Process

When you first connect Debezium to a Postgres table, it follows a well-defined process to capture and stream changes to guarantee consistency:

Initial snapshot (a.k.a backfill)

When starting a new stream on a Postgres table, you can configure Debezium to perform a snapshot of the table. This will add the current state of every row in the table to your Kafka topic as read events.

To complete the snapshot before streaming new changes (all while ensuring no change is missed in the interim), Debezium does the following:

Starts a SELECT transaction
Notes and stores the current LSN in the WAL
Scans the configured tables and generates read events for each row
Commits the SELECT transaction and records the snapshot completion

Continuous streaming

After completing the snapshot, Debezium transitions to streaming changes from the exact WAL position (using the stored LSN) where the snapshot occurred.

As described above, Debezium leverages built-in logical replication to capture these changes from the database with strict ordering.

Event transformation

Debezium transforms PostgreSQL's logical decoding events into Debezium's standardized change event format, which includes:

A consistent structure for inserts, updates, and deletes
The table's primary key as the event key
Both before and after states of the modified row
Metadata about the source transaction and timing

You can also use Single Message Transforms (SMTs - which are a feature of Kafka Connect) to modify message payloads as they flow through the pipeline.

Kafka topic routing

Each table's changes are directed to a dedicated Kafka topic, typically named in the pattern <server-name>.<schema-name>.<table-name>. Once delivered to Kafka, the change is in a durable stream that can be processed by consumers.

Offset tracking

Once a change has been successfully delivered to Kafka, Debezium will periodically update Postgres with the confirmed_flush_lsn associated with the change. Postgres will then remove WAL records from the replication slot.

Importantly, if Debezium fails to deliver messages to Kafka, it won’t notify Postgres it can flush messages, which can cause the replication slot to fill up and could crash the database.

Debezium Postgres Benefits

Complete data capture

Critically, Debezium’s approach to CDC guarantees that every complete insert, update, and delete will be captured from your database. By using WAL records with LSN’s as a cursor, it can provide strict ordering guarantees and is able to enrich messages with transaction metadata.

Non-intrusive

Debezium requires no schema modifications (i.e. no last_updated column) and leverages the WAL ( a normal part of every database transaction) to minimize database overhead.

Consistency

With both snapshots and streaming, Debezium is able to deliver every row in a table to a Kafka topic.

Debezium Postgres Limitations

Complex dependencies

Debezium introduces complex dependencies like Kafka to achieve even simple CDC use cases. This is a significant operational burden.

Replication slot risk

One bad message can lock up Debezium, cause the replication slot to back up, and potentially consume significant resources on the database.

Throughput constraints

Tuning Debezium to match the throughput of your Postgres database is not easy. You’ll need to be comfortable creating multiple replication slots, tuning Debezium beyond its default config, and getting very familiar with Kafka.

Conclusion

Debezium has set the standard for Postgres change data capture. By leveraging log-based replication and the tooling of Kafka Connect, it’s a complete CDC platform.

However, because it isn’t tuned to Postgres and requires Kafka as a dependency, Debezium isn’t easy to use or maintain. A single message can jam your pipeline, so you’ll be spending significant time tuning and monitoring the system (in the JVM!) to keep things running smoothly.

At Sequin, we’re improving on Debezium’s approach - tuning it specifically for Postgres and removing the Kafka dependency. It’s resulting in a 6.8X improvement in speed with higher availability and resiliency - all with nicer tooling.

Choosing the right, real-time, Postgres CDC platform

Eric Goldman — Thu, 17 Apr 2025 02:19:30 +0000

Change Data Capture (CDC) has become a critical component of modern data architectures. Teams use CDC to build event-driven workflows that react to database changes. Or to maintain state across services and data stores.

As a critical component of the stack, teams need a CDC solution that's fast and reliable.

In this guide, we'll compare the leading real-time CDC solutions across three key dimensions that matter most: technical capabilities, required expertise, and budget considerations. As we were building Sequin to address many of the challenges teams face with CDC, we couldn’t find a resource that laid out all the options. Since we did the work to try each tool available today, we thought we’d share our findings.

What is real-time CDC?

In real-time CDC, the moment a row changes in your database (insert, update, delete) you want to deliver that change to another system right away. Usually, you want real-time CDC for operational use cases, not analytics. So a cron job won't do.

You may set up a real-time CDC pipeline for a number of reasons. Those reasons fall into two broad categories:

Event-driven workflows: When you insert, update, or delete a row in your database you want to send an event to one or more services or processors.

Replication: The state of your database needs to be synchronized with other services - be it a database, a cache, search index, third-party API, or a materialized view. Unlike analytics (per above) you need the two systems to be in-sync fast so, for example, a search doesn’t return an unavailable product, etc.

For both kinds of use cases, guarantees are critical. In particular, you want the guarantee of exactly-once processing. These systems are way easier to build and maintain if you know that a downstream service will receive a change when it happens. And that if the downstream service fails to process the change, it will be retried until it succeeds.

The framework

I’ll highlight three characteristics of each platform to help you make the right choice:

Capabilities: I’ll boil this down to three primary considerations that should help you quickly determine if the tool is even close to your needs:

What kinds of changes can the system detect in Postgres?
What kind of deliverability guarantees does the system provide?
What destinations does the system support?

Technical expertise: How easy is the system to set up and maintain? Do you need specific skills to use the tool? How much time will you need to dedicate to it?

Budget constraints: In addition to time, how much will the system cost to operate?

With that, let’s dig in.

Free, open source

There are a handful of open source projects that you can run yourself. You get the benefits of using a widely adopted tool with the ability to tailor it to your needs.

Debezium

https://debezium.io/

TL;DR: Debezium is a widely deployed CDC platform, but it’s notoriously hard to use and requires Kafka expertise.

Capabilities: Debezium captures all inserts, updates, and deletes. Because it uses Kafka, it comes with exactly-once processing guarantees. Debezium's only destination is Kafka, but you can use Kafka Connect to route from Kafka to other tools and services.

Logical replication: inserts, updates, and deletes in real-time
Exactly-once processing
Kafka

Technical expertise: High. Self-hosting Debezium requires deploying and managing the JVM, ZooKeeper, and Kafka. Each requires configuration and ongoing maintenance.

Budget: High. Debezium is free and open source, but the real cost is in complexity. With all the dependencies and configuration the total cost of ownership is significant.

Note: Debezium launched Debezium Server several years ago - which removed the Kafka dependency, and supports more streams/queues. However, it seems pretty basic and has received limited investment - and is hardly being used from what I can tell (only 75 stars on GitHub). It’s worth mentioning, but I don’t see it as a viable offering right now.

Sequin :)

https://sequinstream.com/

TL;DR - Sequin is a new Postgres CDC solution that leverages your existing database infrastructure instead of requiring Kafka. It delivers changes directly to popular streams and queues through native connectors.

Capabilities: Sequin captures every database change using Postgres's native replication capabilities, without requiring additional infrastructure like Kafka. The platform provides exactly-once processing guarantees by leveraging Postgres itself for reliable message delivery. Sequin includes native connectors that deliver changes directly to destinations like Kafka, SQS, NATS, RabbitMQ, and Sequin streams, with built-in transformation capabilities.

Logical replication: inserts, updates, and deletes in real-time
Exactly-once processing guarantees
Native connectors for streams and queues like Kafka, NATS, RabbitMQ, and more

Technical expertise: Low. Sequin runs as a single Docker image. Sequin comes with a web console, CLI, and config files to simplify set up and management. Sequin offers a fully hosted option as well.

Budget: Low. Sequin is a free, open source platform. It’s more cost effective compared to alternatives because it uses your existing Postgres infrastructure.

Hosted

While there are many hosted, closed source CDC providers, few offer real-time data capture with stream or queue destinations. Here’s the set to consider:

Decodable

https://www.decodable.co/

TL;DR - Decodable provides a hosted Debezium plus Flink stack - ideal if you're already invested in these technologies but want them managed.

Capabilities: Decodable provides a hosted version of Debezium paired with Apache Flink. It’ll capture every change in your database with strong deliverability guarantees, monitoring, and schema management tools.

Logical replication: inserts, updates, and deletes in real-time
Exactly-once processing guarantees
Kafka and Pulsar

Technical expertise: Medium to low. The hard part is learning how to configure pipelines and apply SQL transforms in the Decodable dashboard. Flink pipelines definitely come with a learning curve - and the slow feedback loop as you wait for changes to apply doesn’t help.

Budget: Medium. Decodable is fairly expensive. One pipeline can have several different “tasks” that can each cost several hundred dollars per month.

Confluent Debezium

https://docs.confluent.io/kafka-connectors/debezium-postgres-source/current/overview.html

TL;DR - A fully hosted version of Debezium with enterprise features. A reasonable choice if you're already using Confluent.

Capabilities: A fully hosted version of Debezium that comes with enterprise security, compliance, monitoring, and scaling tools built in.

Logical replication: inserts, updates, and deletes in real-time
Exactly-once processing
Kafka

Technical expertise: Medium. While you won’t need to directly manage the deployment, JVM, and dependencies - you’ll still be responsible for the complex configuration of Debezium and Kafka topics

Budget: High, enterprise pricing. You’ll trade engineering time for an expensive enterprise infrastructure product that still requires Kafka expertise.

Confluent Direct JDBC Postgres

https://docs.confluent.io/cloud/current/connectors/cc-postgresql-source.html

TL;DR: If you need Kafka and are paying for Confluent, and you just need to replicate postgres rows to a topic - this is for you!

Capabilities: Unlike every other solution thus far, this connector uses polling to capture changes. This means you won’t capture deletes (which may be a dealbreaker). It’s easier to set up but less powerful.

Polling: inserts and updates only with delay
Exactly-once processing
Kafka

Technical expertise: Medium / low. The Postgres replication is simplified (but less powerful), and your still managing Kafka’s topic configurations, consumer groups, schema evolution, etc.

Budget: Very High. As with other Confluent offerings, you’ll pay a high price for their enterprise tooling. You’ll pay individually for the Postgres connector in addition to your required Kafka instance.

Striim

https://www.striim.com/

TL;DR: Striim is known as an enterprise solution with a high cost - but it delivers a reliable CDC solution. It’s intended for very large companies.

Capabilities: Striim provides CDC with transformations, schema management, delivery guarantees, and security. It's designed for large Fortune 1,000 companies.

Logical replication: inserts, updates, and deletes in real-time
Exactly-once processing
Multiple destinations

Technical expertise: Medium. You'll need to learn Striim's proprietary TQL (Transform Query Language) for data transformations and their StreamApps framework for pipeline configuration. While well-documented, these tools have a steep learning curve and are unique to Striim's ecosystem.

Budget: Very high. Striim requires an all-in contract designed for large enterprises.

Cloud provider tools

The major infrastructure providers offer CDC products that work within their ecosystem. Tools like AWS DMS, GCP Datastream, and Azure Data Factory can be configured to stream changes from Postgres to other infrastructure.

TL;DR: Setting up CDC through your cloud provider can certainly work if you are all-in on one provider and comfortable with their tools.

Capabilities: AWS, GCP, and Azure offer CDC capabilities integrated into their infrastructure. For instance with AWS DMS, you can configure your AWS RDS with CDC to send events to AWS SQS.

WAL: Inserts, Updates, Deletes in Real-time
Variable by configuration and provider
Destinations within the provider

Technical expertise: Medium. Setting up this kind of CDC is all about navigating through your provider’s web console, permissions, tooling, and logging to set up pipelines. You’ll need to be familiar with all the potential settings and ready to guess and check with Claude Sonnet.

Budget: Medium. Often you’ll be paying for extra compute hours and data transfer which can add up quickly and are hard to predict. Additionally, these setups can be brittle and hard to maintain as settings and dependencies are strewn about.

ETL providers (Fivetran, Airbyte, Etc)

If you need real-time CDC, then these platforms are immediately disqualified because they only offer batch ETL. It may take a minute or two for a change in your database to appear in your stream or queue. And even then, depending on your set up, it may not have atomic changes.

That said, they are worth mentioning because they do provide CDC to a handful of streams and queues.

TL;DR: if you just need analytics, these are a good option.

Capabilities: ETL tooling provides schedule, batch delivery of changes in your database to streams like Kafka. It’s primarily intended for non-operational, analytics use cases.

Batch CDC on a schedule
Variable - often at-most-once guarantees.
Multiple destinations

Technical expertise: Low. ETL tools are easy to set up - but not very configurable when used for CDC.

Budget: High. You’ll pay for every row which can get expensive.

Build your own

Of course, you can definitely build this on your own real-time CDC. We’ve written several pieces on this topic to help you get started.

Creating a custom CDC solution offers the ultimate in flexibility and optimization for specific use cases. You can build exactly what they need, with precisely the delivery guarantees and transformation capabilities you require. And Postgres comes with some helpful capabilities to make this relatively easy

However, this path demands the highest level of technical expertise. You’ll be in the weeds of the WAL, Postgres, and some sort of approach to buffer and deliver changes. It’s honestly very fun work - but hard to get right. Especially if you require backfills, strict ordering, exactly-once processing, monitoring, and redundancy.

Conclusion

While CDC is not a new idea, it’s becoming more common as apps become more data intensive. While the options are a little sparse (especially compared to other tooling) - the space is maturing. I’d recommend taking action from here by picking one solution in each category - try an open source solution (like Sequin!), start a free trial of a hosted option, and see if your cloud provider might do the trick. Then you’ll have working knowledge of your option space to make a good decision.

[Boost]

Eric Goldman — Fri, 06 Dec 2024 01:12:49 +0000

Choosing the right, real-time, Postgres CDC platform

Eric Goldman for Sequin ・ Dec 6 '24

#postgres #database #eventdriven #dataengineering

Choosing the right, real-time, Postgres CDC platform

Eric Goldman — Fri, 06 Dec 2024 01:12:25 +0000

As a critical component of the stack, teams need a CDC solution that's fast and reliable.

What is real-time CDC?

You may set up a real-time CDC pipeline for a number of reasons. Those reasons fall into two broad categories:

Event-driven workflows: When you insert, update, or delete a row in your database you want to send an event to one or more services or processors.

The framework

I’ll highlight three characteristics of each platform to help you make the right choice:

Capabilities: I’ll boil this down to three primary considerations that should help you quickly determine if the tool is even close to your needs:

What kinds of changes can the system detect in Postgres?
What kind of deliverability guarantees does the system provide?
What destinations does the system support?

Technical expertise: How easy is the system to set up and maintain? Do you need specific skills to use the tool? How much time will you need to dedicate to it?

Budget constraints: In addition to time, how much will the system cost to operate?

With that, let’s dig in.

Free, open source

There are a handful of open source projects that you can run yourself. You get the benefits of using a widely adopted tool with the ability to tailor it to your needs.

Debezium

TL;DR: Debezium is a widely deployed CDC platform, but it’s notoriously hard to use and requires Kafka expertise.

Logical replication: inserts, updates, and deletes in real-time
Exactly-once processing
Kafka

Sequin

TL;DR: Sequin is a new Postgres CDC solution that leverages your existing database infrastructure instead of requiring Kafka. It delivers changes directly to popular streams and queues through native connectors.

Logical replication: inserts, updates, and deletes in real-time
Exactly-once processing guarantees
Native connectors for streams and queues like Kafka, NATS, RabbitMQ, and more

Budget: Low. Sequin is a free, open source platform. It’s more cost effective compared to alternatives because it uses your existing Postgres infrastructure.

Hosted

While there are many hosted, closed source CDC providers, few offer real-time data capture with stream or queue destinations. Here’s the set to consider:

Decodable

TL;DR: Decodable provides a hosted Debezium plus Flink stack - ideal if you're already invested in these technologies but want them managed.

Logical replication: inserts, updates, and deletes in real-time
Exactly-once processing guarantees
Kafka and Pulsar

Budget: Medium. Decodable is fairly expensive. One pipeline can have several different “tasks” that can each cost several hundred dollars per month.

Confluent Debezium

TL;DR: A fully hosted version of Debezium with enterprise features. A reasonable choice if you're already using Confluent.

Capabilities: A fully hosted version of Debezium that comes with enterprise security, compliance, monitoring, and scaling tools built in.

Logical replication: inserts, updates, and deletes in real-time
Exactly-once processing
Kafka

Budget: High, enterprise pricing. You’ll trade engineering time for an expensive enterprise infrastructure product that still requires Kafka expertise.

Confluent Direct JDBC Postgres

TL;DR: If you need Kafka and are paying for Confluent, and you just need to replicate postgres rows to a topic - this is for you!

Polling: inserts and updates only with delay
Exactly-once processing
Kafka

Technical expertise: Medium / low. The Postgres replication is simplified (but less powerful), and your still managing Kafka’s topic configurations, consumer groups, schema evolution, etc.

Striim

TL;DR: Striim is known as an enterprise solution with a high cost - but it delivers a reliable CDC solution. It’s intended for very large companies.

Capabilities: Striim provides CDC with transformations, schema management, delivery guarantees, and security. It's designed for large Fortune 1,000 companies.

Logical replication: inserts, updates, and deletes in real-time
Exactly-once processing
Multiple destinations

Budget: Very high. Striim requires an all-in contract designed for large enterprises.

Cloud provider tools

TL;DR: Setting up CDC through your cloud provider can certainly work if you are all-in on one provider and comfortable with their tools.

Capabilities: AWS, GCP, and Azure offer CDC capabilities integrated into their infrastructure. For instance with AWS DMS, you can configure your AWS RDS with CDC to send events to AWS SQS.

WAL: Inserts, Updates, Deletes in Real-time
Variable by configuration and provider
Destinations within the provider

ETL providers (Fivetran, Airbyte, Etc)

That said, they are worth mentioning because they do provide CDC to a handful of streams and queues.

TL;DR: if you just need analytics, these are a good option.

Capabilities: ETL tooling provides schedule, batch delivery of changes in your database to streams like Kafka. It’s primarily intended for non-operational, analytics use cases.

Batch CDC on a schedule
Variable - often at-most-once guarantees.
Multiple destinations

Technical expertise: Low. ETL tools are easy to set up - but not very configurable when used for CDC.

Budget: High. You’ll pay for every row which can get expensive.

Build your own

Of course, you can definitely build this on your own real-time CDC. We’ve written several pieces on this topic to help you get started.

Conclusion

How to invoke a lambda function from your database

Eric Goldman — Fri, 27 Sep 2024 20:47:34 +0000

AWS Lambda is a serverless compute service that lets you run code without provisioning or managing servers. It automatically scales your applications in response to incoming requests.

Often, you want to trigger an AWS Lambda function when a database row changes. For example, you may want to trigger a function as a side-effect of a database change, or fan out work to multiple services.

In this guide, you will learn how to setup Sequin to trigger an AWS Lambda function when a database row changes.

Prerequisites

You are about to create a simple AWS Lambda function that logs a message to the console. Sequin will trigger this function by sending a HTTP POST request to the function's URL with the payload of the database row that changed.

You'll need the following:

An AWS account
A Sequin account
A Postgres database (Sequin works with any Postgres database version 12+ - including RDS) with a basic users table containing an id column and a name column.

Create a Lambda function

Start by creating a new AWS Lambda function that takes in a Sequin change event as a payload and logs the payload to the console.

Create a new Lambda function

Open the AWS Lambda console and click "Create function".
Choose "Author from scratch".
Give your function a name (e.g., "newUserHandler").
Select Node.js as the runtime and which ever architecture you want to support (e.g., "arm64").
Click "Create function":

Add function code

Replace the default code in the Lambda function editor with the following:

export const handler = async (event) => {
        // Verify the Sequin webhook secret
        const authHeader = event.headers.authorization;
        if (!authHeader || authHeader !== `Bearer ${process.env.SEQUIN_WEBHOOK_SECRET}`) {
          return {
            statusCode: 401,
            body: JSON.stringify('Unauthorized'),
          };
        }

        try {
          const payload = JSON.parse(event.body);
          const { record } = payload;

          if (record && record.name) {
            console.log(`Hello ${record.name}`);
          } else {
            console.log('No name found in the payload.');
          }

          return {
            statusCode: 200,
            body: JSON.stringify('Success'),
          };
        } catch (error) {
          console.error('Error processing request:', error);
          return {
            statusCode: 500,
            body: JSON.stringify('Internal Server Error'),
          };
        }
      };

This function first checks the authorization header to make sure the request is coming from Sequin. Then it processes the payload (which contains the database row that changed) and logs the name from the payload to the console.

Click Deploy to save the function.

Add the SEQUIN_WEBHOOK_SECRET environment variable

In the Lambda function Configuration tab, scroll down to the "Environment variables" section.
Click "Edit" and then "Add environment variable".
Set the key as SEQUIN_WEBHOOK_SECRET and the value to a secure secret of your choice.
Click "Save".

You will need to use this secret value in the Sequin dashboard when you create the push consumer.

Create a Lambda Function URL

To make your Lambda function accessible via HTTP, you need to create a Function URL:

Go to your Lambda function in the AWS Console.
In the "Configuration" tab, click on "Function URL" in the left sidebar.
Click "Create function URL".
For "Auth type", select "NONE".
Under "Configure cross-origin resource sharing (CORS)", check "Configure CORS".
In the "Allowed origins" field, enter "*" (without quotes) to allow all origins for now. You can restrict this later.
Click "Save".

After saving, you'll see a Function URL. This is the URL you'll use to configure your Sequin consumer in the next section.

Create a Sequin Push Consumer

Create a new Sequin push consumer that detects changes to the users table and sends a HTTP POST request to the Lambda function's URL.

Connect Sequin to your database

Login to your Sequin account and click the Add New Database button.
Enter the connection details for your Postgres database.
Follow the instructions to create a publication and a replication slot by running two SQL commands in your database:

create publication sequin_pub for all tables;
select pg_create_logical_replication_slot('sequin_slot', 'pgoutput');

Name your database and click the Connect Database button.

Sequin will connect to your database and ensure that it's configured properly.

Create a Push Consumer

Create a push consumer that will capture users from your database and deliver them to your Lambda function:

Navigate to the Consumers tab and click the Create Consumer button.
Select your users table (i.e public.users).
For this guide, you want to capture all changes to the users table. To do this, select Changes and click Continue.
Select to capture inserts, updates, and deletes. No need to add a filter for now. Click Continue.
On the next screen, select Push to have Sequin send the events to your webhook URL. Click Continue.
Now, give your consumer a name (i.e. users_push_consumer) and in the HTTP Endpoint section, click New HTTPE Endpoint.
Enter the Lambda Function URL you obtained earlier. Then click to Add Encrypted Header and add an encryption header with the key Authorization and the value Bearer SEQUIN_WEBHOOK_SECRET, using the secret value you set in your Lambda function's environment variables. Click Create Endpoint.
Back in the tab where you were creating your consumer, click the refresh button by the Endpoints section and select the endpoint you just created. Click Create Consumer

Test end-to-end

Create a new user in your database">

insert into
users (name)
values
  (
   'John Doe'
);

Trace the change in the Sequin dashboard

In the Sequin console, open the Messages tab on your consumer and confirm that a message was delivered.

Confirm the event was received by your Lambda function">

Open the AWS Lambda console and navigate to your function.
Click on the "Monitor" tab and then "View CloudWatch logs"
In the most recent log stream, you should see a log entry: Hello John Doe:

Next steps

Modify this example to suit your needs:

Create Lambda functions to perform side-effects, fan out work, and more.
If you need to run long-running jobs, consider using AWS Step Functions in tandem with Lambda functions.
Tune your consumer configuration to suit your volume of work.
Implement additional security measures, such as API Gateway request validation.

All the ways to react to changes in Supabase

Eric Goldman — Wed, 11 Sep 2024 23:45:51 +0000

This video covers all the ways to capture changes in Supabase.

It's a video deep dive into all the topics covered in the original blog post: https://blog.sequinstream.com/all-the-ways...

Read about all the ways to capture changes in Postgres, more generally: https://blog.sequinstream.com/all-the-ways...

Automatically creating Salesforce contacts when a user signs up

Eric Goldman — Thu, 06 Jul 2023 21:23:16 +0000

This guide will show you how to use Sequin to automatically create a Salesforce contact when a user signs up for your app.

Use cases

Creating a new Salesforce Contact when a user signs up for your app is a common use case. It allows you to keep your Salesforce instance up to date with the latest user data so your sales, support, and operations teams can work with reliable customer data.

Building a reliable integration with Salesforce and your application using Sequin comes with several benefits:

Maintainability

Simplifying your integration into a single transaction makes your code easier to maintain. You don't need to track down errors in your client-side code or investigate issues with your data pipeline.

Guarantees

A user missing its paired Salesforce contact is a regular, painful problem. There are many reasons why a user might not be synced to Salesforce, but a very common one is that your client-side code (often provided by an analytics tool like Segment) is being blocked by a browser extension. Or, your pipeline / ETL is hitting an error. By creating a direct server-side integration between Salesforce and your application you couple your logic for creating a user with your logic for creating a Salesforce contact. This ensures that a user is always synced to Salesforce when they are created.

Also, when you create a user and a Salesforce contact in the same transaction, you can ensure that the data in your Salesforce instance is clean and consistent. Any error that would cause drift can be handled in-line. For instance, you can ensure that their email is valid before letting the user move on. You can catch duplicates, or, you can ensure that the user's first name and last name are synced to the contact to create personalization. This is especially important if you are using Salesforce as a source of truth for sales and support communications.

Pre-requisites

This tutorial will assume you've followed the Salesforce setup guide to connect your Salesforce instance and database to Sequin. If you haven't, check out our guide on syncing Salesforce with Sequin. Ensure you are syncing the Contact collection to your database.

This tutorial will use node.js and the node-postgres library to automatically create a Salesforce contact when a user signs up for your app. We'll assume you already have a node project setup with pg installed. If you don't, you can follow the node-postgres getting started guide to get up and running.

Step 1: Setup pg clients

First, create a connection to your database using pg. As a best practice, we recommend configuring two Postgres clients in your application so you can route Salesforce queries through the Sequin proxy and route other queries directly to your database.

Direct client

Configure a pg client that connects directly to your database. Queries through this client will be executed directly against your database without any Sequin dependency. You should use this client for all read and write operations that don't touch your Salesforce data - for instance, creating a user account.

require("dotenv").config();
const { Pool } = require("pg");

const direct = new Pool({
  user: "postgres",
  host: "your.database.host",
  database: "yourDatabaseName",
  password: process.env.DIRECT_DB_PASSWORD,
  port: 5432,
});

Friendly reminder: never store your database credentials in your application code. Use environment variables to store your credentials and load them into your application at runtime. For node.js, you can use a .env file and the dotenv library.

Sequin proxy client

Configure another 'pg' client - but this time configure it to connect to the Sequin Postgres Proxy.

You'll find the connection details for your Sequin proxy in the Sequin console. Select your sync and then navigate to the Connection instructions tab to find your proxy user, host, and password. Then enter them into another pg client configuration:

const salesforceProxy = new Pool({
  user: "sequin_▒▒▒▒▒▒.▒▒▒▒▒▒",
  host: "proxy.sequindb.io",
  database: "yourDatabaseName",
  password: process.env.SALESFORCE_DB_PASSWORD,
  port: 5432,
});

Queries through this client will pass through Sequin to ensure any mutations are applied to Salesforce. You should use this client for write operations that touch your Salesforce data - in this tutorial, creating a new contact.

Step 2: Create contact

Now that you have two pg clients configured, you can write a query to create a new contact in Salesforce.

You'll use the salesforce Postgres client to execute this query so that Sequin can sync the new contact to Salesforce. Here is a simple function that will create a new contact in Salesforce:

const createNewContact = async (user, id) => {
  const client = await salesforceProxy.connect();

  try {
    const res = await client.query(
      `INSERT INTO salesforce.contact (first_name, last_name, email, user_id__c) VALUES ('${user.first_name}', '${user.last_name}', '${user.email}', '${id}');`
    );
    console.log(res);
  } catch (error) {
    console.error("Error creating contact:", error);
    throw new Error(error);
  } finally {
    client.release();
  }
};

This function is designed to be called after a user is created. It takes in the user object and the user's id, opens a connection to the database, and then inserts a new contact into Salesforce. The user_id__c field is a custom field on the Salesforce contact object that will link the Salesforce contact to the user. If the contact is successfully created, the connection to the database is closed. If there is an error - for instance a duplicate user is found - a new error will be thrown. This allows you to handle the error in your application code - perhaps letting the new user know that they already have an account.

Step 3: Create user

Now, incorporate the createNewContact() function into your user creation flow. Here is a simple example of a user creation flow that uses the createNewContact function:

const createNewUser = async (user) => {
  const client = await direct.connect();

  try {
    await client.query("BEGIN");
    const newUser = await client.query(
      "INSERT INTO users (first_name, last_name, email) VALUES ($1, $2, $3) RETURNING id",
      [user.first_name, user.last_name, user.email]
    );
    await createNewContact(user, newUser.rows[0].id);
    await client.query("COMMIT");
    console.log("New user created successfully!");
  } catch (error) {
    await client.query("ROLLBACK");
    console.error("Error creating user:", error);
  } finally {
    client.release();
  }
};

This function is designed to be called when a user signs up for your app. It takes in the user object, opens a connection to the database, and starts a new transaction. It then attempts to create a new user before calling the createNewContact function to create a new contact in Salesforce. If both the user and the contact are successfully created, the transaction is committed. If there is an error - for instance the user's email is invalid - the transaction is rolled back. This ensures that the user and the contact are always created together.

Next steps

You've successfully created a new Salesforce contact when a user signs up for your app. This approach ensures every user will be paired to a Salesforce contact upon creation. It also allows you to leverage your business logic in Salesforce to ensure new users have valid emails and are not duplicated, for example.

From here, you can also use this approach to create other Salesforce objects when a user signs up - for instance, mapping the user to a new or existing Salesforce account. Or automatically creating a new opportunity and assigning it to a sales rep.

Integrate with HubSpot using Prisma

Eric Goldman — Thu, 06 Apr 2023 20:56:17 +0000

Sequin lets you sync your HubSpot data to your database in real-time. You can then use ORMs, like Prisma, to rapidly build your HubSpot integration.

In this playbook, you'll learn how to set up HubSpot to work with Prisma using Sequin. You'll then write your first queries using the Prisma client and explore the development lifecycle as you run migrations and scale your integration.

Starting with an existing project

You're likely already using Prisma as your ORM. So this playbook starts from an existing Prisma project and shows you how to add HubSpot data to your stack.

We assume you've already followed Prisma's quickstart to create your TypeScript project, install Prisma, connect your database, introspect your schema, and query with the Prisma Client.

Specifically, this playbook builds on top of an existing Prisma project connected to a PostgreSQL database with one schema called public containing two tables: users and orgs. Each user is a part of an org — represented in the database with a foreign key relationship:

This schema is represented as a Prisma model in the schema.prisma file:

generator client {
  provider        = "prisma-client-js"
}

datasource db {
  provider = "postgresql"
  url      = env("DATABASE_URL")
}

model orgs {
  id                  Int     @id @default(autoincrement())
  name                String?
  subscription_status String?
  users               users[]
}

model users {
  id         Int     @id @default(autoincrement())
  first_name String?
  last_name  String?
  email      String?
  org_id     Int?
  org        orgs?   @relation(fields: [org_id], references: [id], onDelete: NoAction, onUpdate: NoAction, map: "users_orgs_id_fk")
}

Each table has been translated into a model. And the foreign key relationship between users and orgs has been defined with a @relation scalar.

From the foundation of this existing Prisma project, you'll now add your HubSpot schema to Prisma using Sequin.

Setup your HubSpot sync

To build a sync between HubSpot and your database, Sequin will guide you through the process of authenticating with HubSpot, selecting the data you want to sync, and connecting to your database. Read our HubSpot setup guide for step-by-step instructions.

For the purposes of this playbook, you'll want to sync at least two HubSpot objects mapped together with an association. For instance, you can sync the Contact object, Deal object, and Contact with Deal associations:

To get comfortable with your schema and Prisma's workflows, you also don't need to sync every HubSpot property. In the Sequin Console, configure your sync to include a handful of properties:

For instance, sync some standard properties for the Contact object. For reference, these are the Contact properties used in the remainder of this playbook:

industry
associatedcompanyid
numemployees
website
company
jobtitle
lastname
firstname
user_id

Note: As a helpful example, we include a custom property called user_id, which maps each HubSpot contact to a user. We'll explore this relationship more later in the playbook.

Do the same for the Deals object. Here are the Deal properties used in this playbook:

createdate
closedate
pipeline
dealstage
amount
dealname
hs_priority

Tip: You can also edit and lock the column name for each property you sync. These tools allow you to create simple naming conventions in your database and buffer your database from breaking changes in HubSpot.

With HubSpot configured in Sequin, you can now connect your database to Sequin and create your sync.

After you click Create, Sequin will begin backfilling all the Contacts, Deals, and Contact - Deal associations in your HubSpot instance to your database.

Within a minute, you'll see the new hubspot schema and tables in your database:

You can now configure Prisma to work with your HubSpot data.

Add HubSpot to your Prisma Schema

Next, you'll update your prisma.schema file and then re-generate the Prisma Client to work with the HubSpot data in your database.

Before you do, it's worth building a mental model of this process.

Your database now contains two schemas:

A public schema that you own and control. You're probably used to using Prisma's Migrate tools to make changes to this schema. In Prisma, these are called "model/entity-first migrations." (Some Prisma users prefer to create new tables and update columns in the database via SQL. These are called "database-first migrations.")
A hubspot schema that Sequin owns and controls to maintain your HubSpot sync. All migrations in this schema are done directly to the database via SQL commands that originate from Sequin. So, these are database-first migrations that Sequin runs. If you add or drop columns in this schema using Prisma, you'll break the sync. Hence, all migrations in this schema are performed in the Sequin console. After Sequin applies the migrations to your database, you'll add the changes to Prisma.

So, when working with your hubpost schema, you'll always follow a database-first migration pattern in Prisma. This means you'll pull the schema into Prisma models as opposed to pushing the schema from Prisma models into your database. Here is how.

Turn on multiple schema support

Your database now contains two schemas. To configure Prisma to work with more than one schema, you need to turn on multi schema support.

To do so, update your schema.prisma file as follows:

generator client {
  provider        = "prisma-client-js"
  previewFeatures = ["multiSchema"]
}

datasource db {
  provider = "postgresql"
  url      = env("DATABASE_URL")
  schemas  = ["public", "hubspot"]
}

model orgs {
  id                  Int     @id @default(autoincrement())
  name                String?
  subscription_status String?
  users               users[]

  @@schema("public")
}

model users {
  id         Int     @id @default(autoincrement())
  first_name String?
  last_name  String?
  email      String?
  org_id     Int?
  org        orgs?   @relation(fields: [org_id], references: [id], onDelete: NoAction, onUpdate: NoAction, map: "users_orgs_id_fk")

  @@schema("public")
}

Add previewFeatures = ["multiSchema"] in the generator block. This turns on multi schema support.
List your schemas in the datasource block. In this case: schemas = ["public", "hubspot"]
Designate which schema each of your Prisma models belongs to with a @@schema('public') attribute

Prisma is now ready to handle multiple schemas when you begin to introspect your database in the next step.

Introspect your database

The Prisma CLI provides introspection tools to automatically update your schema.prisma models to reflect the schema in your database.

To do so, ensure you have the Prisma CLI installed, navigate to the root directory of your project, and run the following command:

prisma db pull

Prisma will then retrieve the schema from your database and map your tables, columns, indexes, and constraints into Prisma models, fields, indexes, and attributes in your schema.prisma file.

Caution: If you've manually altered your schema.prisma file, some changes will be over-written when you run db pull. This includes any manually created @relation scalars you've defined in your model. You can instead introspect just your hubspot schema to avoid any issues. Learn how in the Prisma docs.

After introspecting your database, your schema.prisma file will be updated to include your hubspot schema and the underlying contact, deal, and associations_contact_deal models:

generator client {
  provider        = "prisma-client-js"
  previewFeatures = ["multiSchema"]
}

datasource db {
  provider = "postgresql"
  url      = env("DATABASE_URL")
  schemas  = ["hubspot", "public"]
}

model orgs {
  id                  Int     @id @default(autoincrement())
  name                String?
  subscription_status String?
  users               users[]

  @@schema("public")
}

model users {
  id         Int       @id @default(autoincrement())
  first_name String?
  last_name  String?
  email      String?
  org_id     Int?
  org        orgs?     @relation(fields: [org_id], references: [id], onDelete: NoAction, onUpdate: NoAction, map: "users_orgs_id_fk")
  contact    contact[]

  @@schema("public")
}

model associations_contact_deal {
  contact_id       String
  contact          contact? @relation(fields: [contact_id], references: [id])
  deal_id          String
  deal             deal?    @relation(fields: [deal_id], references: [id])
  sync_hash        String?  @map("_sync_hash")
  sync_inserted_at DateTime @default(now()) @map("_sync_inserted_at") @db.Timestamp(6)
  sync_updated_at  DateTime @default(now()) @map("_sync_updated_at") @db.Timestamp(6)
  updated_at       DateTime @default(now()) @map("_updated_at") @db.Timestamp(6)
  labels           String[]

  @@id([contact_id, deal_id], map: "associations_contact_deal_pk")
  @@schema("hubspot")
}

model contact {
  id                        String                      @id(map: "CONTACT_pk")
  sync_inserted_at          DateTime                    @default(now()) @map("_sync_inserted_at") @db.Timestamp(6)
  sync_updated_at           DateTime                    @default(now()) @map("_sync_updated_at") @db.Timestamp(6)
  updated_at                DateTime                    @default(now()) @map("_updated_at") @db.Timestamp(6)
  associatedcompanyid       Decimal?                    @db.Decimal
  company                   String?
  firstname                 String?
  industry                  String?
  jobtitle                  String?
  lastname                  String?
  numemployees              String?
  user_id                   Int?
  user                      users?                      @relation(fields: [user_id], references: [id])
  website                   String?
  associations_contact_deal associations_contact_deal[]

  @@index([updated_at], map: "_updated_at_idx")
  @@schema("hubspot")
}

model deal {
  id                        String                      @id(map: "DEAL_pk")
  sync_inserted_at          DateTime                    @default(now()) @map("_sync_inserted_at") @db.Timestamp(6)
  sync_updated_at           DateTime                    @default(now()) @map("_sync_updated_at") @db.Timestamp(6)
  updated_at                DateTime                    @default(now()) @map("_updated_at") @db.Timestamp(6)
  amount                    Decimal?                    @db.Decimal
  closedate                 DateTime?                   @db.Timestamp(6)
  createdate                DateTime?                   @db.Timestamp(6)
  dealname                  String?
  dealstage                 String?
  hs_priority               String?
  pipeline                  String?
  associations_contact_deal associations_contact_deal[]

  @@schema("hubspot")
}

Define relationships

Out of the box, the schema generated via prisma db pull is almost entirely workable. But because Sequin doesn't enforce foreign key constraints, Prisma can't detect the relationships that exist across your contact and deal tables via the associations_contact_deal relation table. You'll add these relationships to your schema.prisma file.

To define a many-to-many relationship using a relation table, you need to tell Prisma that the deal_id and contact_id fields in the associations_contact_deal model relate to the id field on the deal and contact models respectively. You'll do this by adding the two relation scalars:

model associations_contact_deal {
  contact          contact  @relation(fields: [contact_id], references: [id])
  contact_id       String
  deal             deal     @relation(fields: [deal_id], references: [id])
  deal_id          String
  sync_hash        String?  @map("_sync_hash")
  sync_inserted_at DateTime @default(now()) @map("_sync_inserted_at") @db.Timestamp(6)
  sync_updated_at  DateTime @default(now()) @map("_sync_updated_at") @db.Timestamp(6)
  updated_at       DateTime @default(now()) @map("_updated_at") @db.Timestamp(6)
  labels           String[]

  @@id([contact_id, deal_id], map: "associations_contact_deal_pk")
  @@schema("hubspot")
}

Additionally, on the deal and contact models, you need to define the other side of this many-to-many relationship by pointing a new field back to the associations_contact_deal model:

model deal {
  id               String                      @id(map: "DEAL_pk")
  sync_inserted_at DateTime                    @default(now()) @map("_sync_inserted_at") @db.Timestamp(6)
  sync_updated_at  DateTime                    @default(now()) @map("_sync_updated_at") @db.Timestamp(6)
  updated_at       DateTime                    @default(now()) @map("_updated_at") @db.Timestamp(6)
  amount           Decimal?                    @db.Decimal
  closedate        DateTime?                   @db.Timestamp(6)
  createdate       DateTime?                   @db.Timestamp(6)
  dealname         String?
  dealstage        String?
  hs_priority      String?
  pipeline         String?
  contacts         associations_contact_deal[]

  @@schema("hubspot")
}

model contact {
  id                  String                      @id(map: "CONTACT_pk")
  sync_inserted_at    DateTime                    @default(now()) @map("_sync_inserted_at") @db.Timestamp(6)
  sync_updated_at     DateTime                    @default(now()) @map("_sync_updated_at") @db.Timestamp(6)
  updated_at          DateTime                    @default(now()) @map("_updated_at") @db.Timestamp(6)
  associatedcompanyid Decimal?                    @db.Decimal
  company             String?
  firstname           String?
  industry            String?
  jobtitle            String?
  lastname            String?
  numemployees        String?
  user_id             Int?
  website             String?
  deals               associations_contact_deal[]

  @@index([updated_at], map: "_updated_at_idx")
  @@schema("hubspot")
}

As you may recall, you can also define a one-to-one relationship between the user_id field in the contact model with the id field on the users model. This will make querying across your internal data model and HubSpot data possible. To do so, add one more relation scalar to your contact model:

model contact {
  id                  String                      @id(map: "CONTACT_pk")
  sync_inserted_at    DateTime                    @default(now()) @map("_sync_inserted_at") @db.Timestamp(6)
  sync_updated_at     DateTime                    @default(now()) @map("_sync_updated_at") @db.Timestamp(6)
  updated_at          DateTime                    @default(now()) @map("_updated_at") @db.Timestamp(6)
  associatedcompanyid Decimal?                    @db.Decimal
  company             String?
  firstname           String?
  industry            String?
  jobtitle            String?
  lastname            String?
  numemployees        String?
  user_id             Int?
  user                users?                      @relation(fields: [user_id], references: [id])
  website             String?
  deals               associations_contact_deal[]

  @@index([updated_at], map: "_updated_at_idx")
  @@schema("hubspot")
}

Generate your Prisma Client

All the relationships in your data model are now defined in your schema.prisma file. The last step before writing your first query is to re-generate your Prisma client:

prisma generate

Query HubSpot using the Prisma Client

Prisma gives you a modern, intuitive API for querying your HubSpot data. For instance, you can return all your HubSpot deals in one simple query:

```js index.js
async function get_all_deals() {
const deals = await prisma.deal.findMany();

console.log(deals);
}




No pagination. No authentication token. No metering your requests through a rate limiter.

Or, you can return all the deals at a certain stage as well as all the contacts associated with those deals:



```js index.js
async function get_qualified_deals() {
  const deals = await prisma.deal.findMany({
    where: {
      dealstage: "qualifiedtobuy",
    },
    include: {
      contacts: {
        include: {
          contact: true,
        },
      },
    },
  });

  console.log(deals);
}

You'll note that in your IDE, you get helpful type warnings and type-ahead support as you write these queries—niceties missing from the HubSpot API and SDK.

More impactful to your productivity, you can query HubSpot and your internal data together. For instance, you can query for all the deals associated to a specific user in your public schema:

async function get_user_deals() {
  const user_deals = await prisma.users.findUnique({
    where: { email: eric@sequin.io },
    include: {
      contact: {
        include: {
          deals: {
            include: {
              deal: true,
            },
          },
        },
      },
    },
  });

  console.log(user_deals);
}

This query builds on the relationships you defined in your schema.prisma file to return the deals related to eric@sequin.io. In one Prisma query, you do the work of a SQL query paired with three nested calls to the HubSpot API.

Migrations

Inevitably, the data you need from HubSpot will change, and you'll need to migrate your Prisma schema. While a more comprehensive guide to migrations should be referenced to avoid downtime in your application, here are the order of operations to consider when using Sequin with Prisma.

As noted above, your hubspot schema is managed by Sequin. For simplicity, all migrations start in the Sequin Console and are applied to Prisma as database-first migrations. As a result, you will not use Prisma Migrate when making changes to the HubSpot objects and properties syncing to your database.

Here are some common scenarios:

Adding or removing a HubSpot property

Step 1: If you're about to remove a HubSpot property, first remove it from your app and your Prisma Client.

Step 2: Add/remove the property to/from your sync by editing your table and column mapping in the Sequin Console.

When you click Update, Sequin will immediately migrate your database to add the new property as a new column or drop the column related to the property you removed. In the case where you added a new property, Sequin will begin the backfill process, syncing the property to your database.

Step 3: If you are adding a new property, you'll now manually add the field associated with the new column/property in the appropriate model in your schema.prisma file. It is easier to do this by hand as opposed to introspecting your database so that you can preserve the other relationship scalars.

Step 4: Run the prisma generate command in your terminal to update the Prisma Client.

Adding or removing a HubSpot Object

Step 1: If you're about to remove a HubSpot object, first remove it from your app and your Prisma Client.

Step 2: Add/remove the object to/from your sync by editing your table and column mapping in the Sequin Console.

When you click Update, Sequin will run a migration to create or drop the tables related to your change.

Step 3: If you are adding a new HubSpot object to your sync, you'll now add these new tables as models in your schema.prisma file. You have two options:

Manually add or remove the appropriate models.
⚠️ Partial Introspect Workaround ⚠️: Partially introspect your database by first saving your existing schema.prisma file. Then follow Prisma's guide for introspecting just a subset of your database. In this case, you'll create a new database user that can only access the new tables added to your database by Sequin. Then, append the new models generated by your introspection to your saved prisma.schema file. This is often a more straightforward path when adding an object to your HubSpot sync.

Step 4: Run the prisma generate command in your terminal to update the Prisma Client.

Development lifecycle

You aren't using Prisma Migrate to track the migrations related to changes in your hubspot schema and the underlying Sequin sync. So how does this all work across dev and staging environments?

To support development and staging environments, you'll set up a dev HubSpot account and a dev Sequin sync. You'll then use Sequin's copy config setting to copy changes from the dev sync and database to your production sync and database. Here is the flow.

Step 1: Setup a HubSpot development account to create a new HubSpot instance you can use for development.

Step 2: Sync your HubSpot development account as a new sync in Sequin. Follow the steps at the start of this guide.

Step 3: Make changes and migrations using this development account and sync.

Step 4: When you are ready to deploy these changes to production, copy the changes from your development sync in Sequin to your production sync:

First, make sure any new properties or settings in your dev HubSpot instance have been applied to your production instance.
Then, click to edit your production sync in Sequin. In the Select tables and columns section, click the Copy config from... button. Select your dev sync to apply your development sync configuration to your production sync.
Click Update.

Sequin will run a migration to apply any schema changes.

Step 5: In Prisma, update your schema.prisma file to reflect your dev environment, then run the prisma generate command to update your client.