Debezium CDC Connector (GCP Cloud SQL PostgreSQL)

You want every insert, update, and delete in a PostgreSQL database to appear as
a Kafka message, without your application having to publish those events
itself. The Debezium PostgreSQL connector does this by reading the database's
logical replication stream.

This guide covers GCP Cloud SQL PostgreSQL as the source database
specifically; the database preparation steps differ on other PostgreSQL
hosting. The connector runs on your Kafka Connect cluster and is managed
through the Kafka Connector (app-k8s-kafka-connector) Component in SVIEW.
For setting up the Connect cluster itself, see BigQuery Connector.

If something goes wrong, see Debezium CDC Troubleshooting.

Prerequisites

Enable logical replication on the database

In the Cloud SQL Console, open Edit instance → Flags and set:

cloudsql.logical_decoding = on

The instance restarts. Afterwards confirm the setting took effect:

SHOW wal_level;  -- must return 'logical'

Point Debezium at the primary instance. Cloud SQL read replicas use
physical streaming replication, so logical replication slots cannot be created
on them.

Create a replication user

Create a dedicated database user for Debezium and grant it what it needs:

ALTER USER debezium WITH REPLICATION;
GRANT USAGE ON SCHEMA <myschema> TO debezium;
GRANT SELECT ON ALL TABLES IN SCHEMA <myschema> TO debezium;
CREATE PUBLICATION debezium_pub FOR ALL TABLES;

1. Wire up the database credentials

The connector reads the database password from a secret that the platform
mounts into the Kafka Connect pods. Add the dependency.db block to both
the Kafka Connect and the Kafka Connector Component values:

dependency:
  db:
    type: csql-postgres        # GCP Cloud SQL PostgreSQL
    instance: <instance-name>
    namespace: <db-namespace>
    user: debezium
    keys:
      - secretKey: username
        envKey: DB_USERNAME
      - secretKey: password
        envKey: DB_PASSWORD
      - secretKey: endpoint
        envKey: DB_HOST

The values land as files in
/opt/kafka/external-configuration/db-credentials/. Reference them from the
connector configuration with DirectoryConfigProvider:

database.password: "${dir:/opt/kafka/external-configuration/db-credentials:DB_PASSWORD}"

Mind the syntax. In ${dir:<directory>:<filename>} the directory part ends
at db-credentials; DB_PASSWORD is the filename after the second colon, not
part of the path. Getting this wrong is the most common cause of an
authentication failure. DirectoryConfigProvider itself is enabled by default
in the Kafka Connect Component, so you do not need to configure it.

2. Grant the Kafka ACLs

The Kafka Connect user needs ACLs on your CDC topic prefix and on the
Debezium heartbeat topic. Either grant the default heartbeat topic explicitly
in your Kafka ACL Component:

rawAcls:
  - topic: __debezium-heartbeat.<topic.prefix>
    role: admin
    user: debezium

Or move the heartbeat topic under your existing prefix so the ACLs you already
have cover it:

heartbeat.topics.prefix: "<topic.prefix>.heartbeat"

See Access Management for how ACLs work.

3. Define the connector

A complete example for the Kafka Connector Component:

serviceAccount:
  create: true

dependency:
  db:
    type: csql-postgres
    instance: <instance-name>
    namespace: <db-namespace>
    user: debezium
    keys:
      - secretKey: username
        envKey: DB_USERNAME
      - secretKey: password
        envKey: DB_PASSWORD
      - secretKey: endpoint
        envKey: DB_HOST

connectors:
  - name: cdc-mydb-psql
    class: io.debezium.connector.postgresql.PostgresConnector
    tasksMax: 1
    autoRestart:
      enabled: true
      # omit maxRestarts for unlimited retries — recommended for CDC

    config:
      database.hostname: <db-host>
      database.port: "5432"
      database.user: debezium
      database.password: "${dir:/opt/kafka/external-configuration/db-credentials:DB_PASSWORD}"
      database.dbname: mydb

      # All CDC topics are named <topic.prefix>.<schema>.<table>
      topic.prefix: <stage>.<tenant>.debezium

      plugin.name: pgoutput
      slot.name: debezium_mydb
      publication.name: debezium_pub

      # Tables to capture, schema.table format, comma-separated
      table.include.list: myschema.shop_order

      snapshot.mode: initial

      key.converter: org.apache.kafka.connect.json.JsonConverter
      key.converter.schemas.enable: "false"
      value.converter: org.apache.kafka.connect.json.JsonConverter
      value.converter.schemas.enable: "false"

      # Keeps the replication slot alive during quiet periods
      heartbeat.interval.ms: "10000"
      heartbeat.topics.prefix: "<stage>.<tenant>.debezium.heartbeat"

      # Topic auto-creation, required because brokers do not auto-create
      topic.creation.default.replication.factor: "3"
      topic.creation.default.partitions: "3"

Snapshot modes

Mode	Behaviour
`initial`	Snapshot existing rows, then stream. The default, and safe for a new connector.
`never`	Skip the snapshot and stream from the current WAL position.
`always`	Re-snapshot on every restart. Avoid in production.
`initial_only`	Snapshot only, no streaming. For one-off bulk loads.

For the topics a connector creates and the shape of each change event, see Debezium Output Topics and Event Format.