PostgreSQL CDC Source Connector
Stream database change logs over logical decoding (pgoutput) directly into Kafka topics via krabka-connect-postgres.
krabka-connect framework and its PostgreSQL CDC connector are currently in active development. Logical decoding and checkpoint resumption are exercised by a Docker-backed acceptance suite. The worker runs one connector per process.
krabka-connect-postgres captures row-level inserts, updates, and deletes directly from PostgreSQL write-ahead logs using the native pgoutput logical replication stream. Each record carries the source LSN, transaction id, and commit timestamp, and DELETE events become tombstones. The connector registers its Protobuf key and value schemas with a Confluent-compatible Schema Registry at startup, and it needs no JVM sidecar. It does not take an initial snapshot: it emits only the changes that logical decoding makes available.
Connector Architecture
The connector framework is split across five modular crates in the krabka-connect repository:
| Crate | Role & Responsibility |
|---|---|
| krabka-connect | The connector SPI: Source, Sink, and Converter traits with checkpoint contracts. |
| krabka-connect-derive | Derive macros for declarative connector configuration schemas (#[derive(ConnectorConfig)]). |
| krabka-connect-postgres | PostgreSQL CDC source connector (PostgresWalSource) over pgoutput replication slots. |
| krabka-connect-worker | Connector worker runtime: polls sources, batches into Kafka producers, and checkpoints offsets. |
| krabka-replicator | Cluster-to-cluster replication engine wire-compatible with Kafka MirrorMaker 2. |
Running the Worker
krabka-connect-worker is configured by flags or environment variables. It creates the replication slot, and the publication for the configured tables, when they are missing. Records go to <topic-prefix>.<schema>.<table>, for example db.public.orders. The library-level PostgresSourceConfig keys are database_url, schema_registry_url, slot_name, publication_name, schema, tables, and max_messages_per_poll:
# Each krabka-connect-worker process runs one connector. Every flag also has a
# KRABKA_* environment variable; the connection URL is a secret, so pass it there.
export KRABKA_POSTGRES_URL='postgres://cdc_user:[email protected]:5432/inventory'
krabka-connect-worker \
--connector-id postgres-inventory-cdc \
--kafka-bootstrap kafka:9092 \
--schema-registry-url http://schema-registry:8081 \
--postgres-slot krabka_inventory_slot \
--postgres-publication inventory_publication \
--postgres-schema public \
--postgres-tables orders,line_items,customers \
--topic-prefix db
On Kubernetes, the operator runs the same worker from a KafkaConnector resource (the krabka.io/cluster label is required, and the database URL comes from a Secret key). The operator chart pins the worker image to ghcr.io/krabka-io/krabka-connect-worker:v0.4.1; set spec.image to use another tag or your own build:
apiVersion: krabka.io/v1alpha1
kind: KafkaConnector
metadata:
name: postgres-inventory-cdc
labels:
krabka.io/cluster: demo
spec:
type: postgresSource
databaseUrl:
name: inventory-db
key: url
schemaRegistryUrl: http://schema-registry:8081
slot: krabka_inventory_slot
publication: inventory_publication
schema: public
tables: [orders, line_items, customers]
topicPrefix: dbCheckpoints and delivery
The worker writes with an idempotent producer and acks=all, and stores the acknowledged Log Sequence Number (LSN) in the compacted __krabka_connect_offsets topic. It advances the PostgreSQL slot only after both the Kafka write and the checkpoint are durable. On restart, the worker resumes from the checkpoint. Delivery is at least once: a crash between the two writes can replay a batch of records.
# The Postgres CDC acceptance suite needs Docker. It starts Postgres and a schema
# registry in containers, runs the worker, force-kills it, restarts it, and checks
# that the records still arrive after the restart, including across an ALTER TABLE.
cargo test -p krabka-connect-worker --test postgres_cdc -- --ignored