Connect Cdc Spanner

Redpanda Connect CDC: Google Cloud Spanner

The gcp_spanner_cdc input in Redpanda Connect streams change data capture (CDC) from a Google Cloud Spanner database into Redpanda or any Kafka-compatible topic. It uses Spanner's native change stream API, tracks multiple concurrent partitions, persists watermark state in a Spanner metadata table, and delivers each row mutation as a JSON message with rich metadata.

Introduced in version 4.56.0. This is an Enterprise feature — a Redpanda Enterprise license is required.

The connector supports both GoogleSQL and PostgreSQL Spanner dialects. It automatically detects the dialect and creates the metadata table in the same database if it does not already exist. Partitions split and merge over time as Spanner scales; the connector handles all partition lifecycle events transparently.

Quickstart

1. Create the Spanner change stream (one DDL statement)

-- Track all tables in the database:
CREATE CHANGE STREAM AllChanges FOR ALL;

-- Or track specific tables:
CREATE CHANGE STREAM OrderChanges FOR orders, customers;

-- Spanner default is OLD_AND_NEW_VALUES; opt into NEW_VALUES to drop old values:
CREATE CHANGE STREAM OrderChanges FOR orders, customers
  OPTIONS (value_capture_type = 'NEW_VALUES');

2. Create a GCP service account and grant it the required IAM roles

The connector issues a CREATE TABLE IF NOT EXISTS DDL statement on every startup to create or validate the partition metadata table. This means the service account must retain DDL permission (spanner.databases.updateDdl) permanently — not just on first run.

# Create a service account
gcloud iam service-accounts create redpanda-spanner-cdc \
  --display-name="Redpanda Spanner CDC"

PROJECT=MY_PROJECT
SA=redpanda-spanner-cdc@${PROJECT}.iam.gserviceaccount.com

# Grant Spanner Database Reader (read data and change streams)
gcloud spanner databases add-iam-policy-binding my-database \
  --instance=my-spanner-instance \
  --project=${PROJECT} \
  --member="serviceAccount:${SA}" \
  --role="roles/spanner.databaseReader"

# Grant DDL permission — required on every startup for metadata table setup
gcloud spanner databases add-iam-policy-binding my-database \
  --instance=my-spanner-instance \
  --project=${PROJECT} \
  --member="serviceAccount:${SA}" \
  --role="roles/spanner.databaseAdmin"

# Download the JSON key file and base64-encode it for credentials_json
gcloud iam service-accounts keys create spanner-cdc-key.json \
  --iam-account=${SA}

export SPANNER_CDC_CREDENTIALS=$(base64 < spanner-cdc-key.json)

3. Full pipeline YAML

# spanner-cdc-pipeline.yaml
input:
  label: "spanner_cdc"
  gcp_spanner_cdc:
    # Base64-encoded service account JSON, or leave empty to use ADC
    credentials_json: "${SPANNER_CDC_CREDENTIALS}"
    project_id: "my-gcp-project"
    instance_id: "my-spanner-instance"
    database_id: "my-database"
    stream_id: "OrderChanges"
    # Optional: start from a specific point in time (RFC3339)
    # start_timestamp: "2025-01-01T00:00:00Z"
    # Optional: stop at a specific time (RFC3339, exclusive)
    # end_timestamp: "2025-12-31T23:59:59Z"
    heartbeat_interval: 10s
    # metadata_table defaults to: cdc_metadata_OrderChanges
    # allowed_mod_types filters to only these operations:
    allowed_mod_types:
      - INSERT
      - UPDATE
      - DELETE
    batching:
      count: 100
      period: 1s

output:
  kafka_franz:
    seed_brokers:
      - "localhost:9092"
    topic: '${! meta("table_name") }'
    compression: snappy

4. Run the pipeline

# Assuming redpanda-connect binary is in PATH (Enterprise build)
redpanda-connect run spanner-cdc-pipeline.yaml

# Or with rpk (Redpanda enterprise)
rpk connect run spanner-cdc-pipeline.yaml

5. Inspect a message

Each emitted message payload is a JSON-serialized Mod object:

{
  "keys":       { "SingerId": "1" },
  "new_values": { "FirstName": "Alice", "LastName": "Smith" },
  "old_values": {}
}

Message metadata keys available via meta("..."):

How It Works

The connector uses the Spanner Change Stream API, which divides the stream into partitions — each covering an immutable key range for a specific time window. The connector:

1. Creates a metadata table in the same Spanner database (default name: cdc_metadata_<stream_id>) to track partition state (CREATED → SCHEDULED → RUNNING → FINISHED) and per-partition watermarks.
2. Discovers root partitions at startup (or on resume).
3. Queries each partition concurrently for change records.
4. When a partition splits or merges, the connector automatically detects and schedules child partitions.
5. Watermarks are updated after each message is acknowledged downstream.
6. On restart, interrupted (SCHEDULED or RUNNING) partitions are resumed from their last watermark.

Reference Directory

• Config Reference: Every gcp_spanner_cdc field with type, default, and description — grounded in the source.
• Setup Spanner: Creating the change stream, IAM permissions, metadata table details, dialects, and retention notes.
• Pipeline and Output: Full pipeline YAML examples, the message payload and metadata shape, per-table routing with Bloblang, landing CDC into an Iceberg topic, and restart/resume behavior.
• Enterprise Features: Redpanda enterprise differentiators for Spanner CDC — the connector's Enterprise license gating; Iceberg Topics (redpanda.iceberg.mode/delete/invalid.record.action/partition.spec/target.lag.ms); Tiered Storage (redpanda.remote.write/read, cloud_storage_enabled); Cloud Topics (redpanda.cloud_topic.enabled / redpanda.storage.mode=cloud, cloud_topics_enabled); Remote Read Replicas (redpanda.remote.readreplica); Shadowing DR (rpk shadow); the Connect redpanda config-service block, secrets management, allow/deny lists, FIPS; and destination-cluster security (RBAC, OIDC/OAUTHBEARER, Kerberos, Audit Logging, Schema ID Validation). All require an Enterprise license.