Postgres-to-NATS CDC with Debezium

application.properties

# Debezium 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=postgres
debezium.source.database.port=5432
debezium.source.database.user=postgres
debezium.source.database.password=postgres
debezium.source.database.dbname=postgres
debezium.source.topic.prefix=events
# comma-separated list of tables to include in monitoring
debezium.source.table.include.list=public.eventoutbox

# Debezium sink configuration for NATS JetStream
debezium.sink.type=nats-jetstream
debezium.sink.nats-jetstream.url=nats://nats:4222
debezium.sink.nats-jetstream.create-stream=true
debezium.sink.nats-jetstream.subjects=event.*
debezium.sink.nats-jetstream.storage=memory

# Event Router SMT configuration
debezium.transforms=outbox
debezium.transforms.outbox.type=io.debezium.transforms.outbox.EventRouter
debezium.transforms.outbox.route.topic.replacement=event.routedByValue
debezium.transforms.outbox.table.expand.json.payload=true

docker-compose.yaml

version: "3.9"
services:
  postgres:
    image: docker.io/debezium/postgres:16-alpine
    ports:
      - "5432:5432"
    environment:
      POSTGRES_PASSWORD: postgres
    volumes:
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 2s
      timeout: 2s
      retries: 5

  nats:
    image: docker.io/nats:2.10.5
    ports:
      - "4222:4222"
      - "8222:8222"
    command:
      - "--debug"
      - "--http_port=8222"
      - "--js"
      - "-DV"

  debezium:
    image: docker.io/debezium/server:2.5.4.Final
    volumes:
      - ./application.properties:/debezium/conf/application.properties
    depends_on:
      postgres:
        condition: service_healthy
      nats:
        condition: service_started

init.sql

CREATE TABLE
  eventoutbox (
    id uuid NOT NULL PRIMARY KEY,
    aggregatetype character varying(255) NOT NULL,
    aggregateid character varying(255) NOT NULL,
    type character varying(255) NOT NULL,
    payload jsonb NOT NULL
  );