Connect Cdc Sqlserver

Redpanda Connect CDC: Microsoft SQL Server

The microsoft_sql_server_cdc input in Redpanda Connect streams change events from Microsoft SQL Server's native CDC change tables directly into Redpanda or any Kafka-compatible output. It reads SQL Server's built-in change tables (cdc.<schema>_<tablename>_CT) using LSN (Log Sequence Number) ordering, optionally precedes streaming with a consistent point-in-time snapshot of existing rows, and durably checkpoints the last-delivered LSN so the pipeline can resume after a restart without re-consuming stale data.

This is an Enterprise component — it requires a valid Redpanda Enterprise license. The component name registered in Connect is microsoft_sql_server_cdc.

Quickstart

Step 1–3: Run the following T-SQL in SSMS or sqlcmd (as a sysadmin user):

-- 1. Enable CDC on the database
USE MyDatabase;
EXEC sys.sp_cdc_enable_db;
GO

-- 2. Enable CDC on each table you want to capture
EXEC sys.sp_cdc_enable_table
  @source_schema = N'dbo',
  @source_name   = N'orders',
  @role_name     = NULL;
GO

EXEC sys.sp_cdc_enable_table
  @source_schema = N'dbo',
  @source_name   = N'customers',
  @role_name     = NULL;
GO

-- 3. Create the rpcn schema for the built-in checkpoint cache
--    and grant the Connect user rights to create objects in it
CREATE SCHEMA rpcn;
GO

GRANT CREATE TABLE TO connect_user;
GRANT CREATE PROCEDURE TO connect_user;
GRANT ALTER ON SCHEMA::rpcn TO connect_user;
GO

Step 4: Run the pipeline from a shell:

rpk connect run mssql-cdc.yaml

mssql-cdc.yaml (copy-pasteable):

input:
  microsoft_sql_server_cdc:
    connection_string: "sqlserver://connect_user:secret@sqlserver-host/MyDatabase?database=MyDatabase"
    stream_snapshot: true          # snapshot existing rows before streaming changes
    max_parallel_snapshot_tables: 2
    snapshot_max_batch_size: 1000
    include:
      - "^dbo\\.orders$"      # anchored for exact match — unanchored also matches dbo.orders_archive
      - "^dbo\\.customers$"
    # Built-in checkpoint cache: auto-creates rpcn.CdcCheckpointCache table
    checkpoint_cache_table_name: rpcn.CdcCheckpointCache
    checkpoint_limit: 1024
    stream_backoff_interval: 5s
    batching:
      count: 100
      period: 1s

output:
  kafka_franz:
    seed_brokers:
      - localhost:9092
    topic: '${! meta("table") }'   # route each table to its own topic
    key: '${! json("id") }'        # key by primary key for correct ordering and log compaction

To route to a Redpanda Cloud cluster with SASL:

output:
  kafka_franz:
    seed_brokers:
      - seed-abc123.cloud.redpanda.com:9092
    tls:
      enabled: true
    sasl:
      - mechanism: SCRAM-SHA-256
        username: my-user
        password: "${REDPANDA_PASSWORD}"
    topic: '${! "mssql." + meta("database_schema") + "." + meta("table") }'

Prerequisites

Before starting the pipeline:

1. SQL Server CDC enabled at database level — sys.sp_cdc_enable_db must have been run on the source database.
2. SQL Server Agent running — CDC uses Agent jobs to scan the transaction log and populate change tables. Without it, change tables stay empty.
3. Capture instance per table — sys.sp_cdc_enable_table creates a change table (cdc.<schema>_<tablename>_CT) for each table. Only tables with a capture instance emit changes.
4. Table has a primary key — the snapshot phase issues keyset-pagination queries; tables without a primary key cannot be snapshotted and will error at connect time.
5. Connect user permissions — the user in connection_string needs SELECT on the source tables and the CDC change tables. For the built-in checkpoint cache it also needs CREATE TABLE and CREATE PROCEDURE rights on the target schema, and that schema (rpcn by default) must already exist.

See setup-sqlserver.md for the complete T-SQL setup walkthrough.

Config Fields

All fields are under input.microsoft_sql_server_cdc.

See config-reference.md for full per-field detail including validation rules for checkpoint_cache_table_name.

Emitted Message Shape

Each message body is a JSON object containing the changed row's column values. System columns (__$start_lsn, __$operation, etc.) are stripped; only user-defined columns are present.

{
  "id": 42,
  "customer_name": "Acme Corp",
  "amount": "199.99",
  "created_at": "2024-03-15T10:30:00Z"
}

Metadata fields set on change messages:

The lsn metadata is the raw binary representation of the SQL Server LSN, not the 0x… hex form you see in log output (which uses .String()). To use it in a Bloblang expression or Kafka header as a readable value, hex-encode it first, for example:

pipeline:
  processors:
    - mutation: |
        root = this
        root._lsn_hex = meta("lsn").encode("hex")

Snapshot (operation: read) rows have no lsn metadata key — they are built with LSN: nil in the snapshot phase and are not individually checkpointed.

Operation Types

Updates produce two consecutive messages: update_before followed by update_after with the same LSN. Use a Bloblang processor or filter to keep only one if needed.

Snapshot Behaviour

When stream_snapshot: true and no prior LSN checkpoint exists, the pipeline:

1. Captures the current max LSN via sys.fn_cdc_get_max_lsn().
2. Snapshots each included table in parallel (up to max_parallel_snapshot_tables) using SNAPSHOT isolation transactions and keyset pagination.
3. Emits all existing rows with operation: read.
4. Stores the captured LSN as the checkpoint.
5. Begins streaming from that LSN in the change tables.

If a checkpoint already exists (restart), the snapshot phase is skipped regardless of stream_snapshot.

Checkpointing

By default, the pipeline auto-creates a stored procedure and then a table under the rpcn schema in the source database (in this order):

1. Stored procedure: rpcn.CdcCheckpointCacheUpdate (upsert, created first via CREATE OR ALTER PROCEDURE)
2. Table: rpcn.CdcCheckpointCache (columns: cache_key varchar(7), cache_val varchar(100), created if absent)

The rpcn schema must already exist before starting the pipeline — the pipeline creates the table and procedure but not the schema. Grant the Connect user CREATE TABLE and CREATE PROCEDURE on this schema.

To use an external Connect cache resource instead:

cache_resources:
  - label: my_redis_cache
    redis:
      url: redis://localhost:6379

input:
  microsoft_sql_server_cdc:
    connection_string: "sqlserver://user:pass@host/MyDB"
    checkpoint_cache: my_redis_cache
    checkpoint_cache_key: mssql_orders_cdc
    include:
      - "dbo\\.orders"

Column Type Mapping

The connector has two distinct code paths with different type handling: the snapshot mapper (snapshot.go prepSnapshotScannerAndMappers) and the stream mapper (stream.go mapScannedValue). The types below reflect what is emitted into the message body; the schema metadata uses Benthos common schema type names independent of the path.

Snapshot path (operation: read)

Stream path (operation: insert / update_before / update_after / delete)

The stream mapper only special-cases decimal/money types; all other values come directly from the go-mssqldb driver:

The snapshot and stream paths use different mappers, so the emitted type for a given column may differ between an initial snapshot row and a subsequent CDC change event. Treat downstream consumers for non-decimal types defensively (accept both string and native forms).

Per-Table Topic Routing

Use Bloblang in the output topic field or a switch processor:

output:
  kafka_franz:
    seed_brokers:
      - localhost:9092
    topic: '${! "sqlserver." + meta("database_schema") + "." + meta("table") }'

Or use a switch output for explicit routing:

output:
  switch:
    cases:
      - check: 'meta("table") == "orders"'
        output:
          kafka_franz:
            seed_brokers: [localhost:9092]
            topic: orders-cdc
      - check: 'meta("table") == "customers"'
        output:
          kafka_franz:
            seed_brokers: [localhost:9092]
            topic: customers-cdc

License

microsoft_sql_server_cdc is an Enterprise-licensed Redpanda Connect connector. Attempting to run without a license produces an error at startup; after the 30-day evaluation period, enterprise connectors are blocked unless you upgrade.

Apply the Connect license in any of these ways (grounded in connect/internal/cli/flags_redpanda.go and connect/internal/license/service.go):

• --redpanda-license <string> flag on rpk connect run / rpk connect dry-run (takes precedence).
• REDPANDA_LICENSE env var (inline license string).
• REDPANDA_LICENSE_FILEPATH env var (path to a license file).
• Default file /etc/redpanda/redpanda.license (auto-applied if present and none of the above are set).

This license authorizes the connector. The Redpanda cluster you write into needs its own valid license to use destination-side enterprise features (Iceberg, Schema ID Validation, Tiered Storage, Cloud Topics).

Destination Enterprise Features

When the CDC pipeline writes into Redpanda, these Enterprise (cluster-licensed) features pair directly with it. Full nested keys are in enterprise-features.md.

• Iceberg Topics (lakehouse landing of CDC events): cluster iceberg_enabled; per-topic redpanda.iceberg.mode (key_value / value_schema_id_prefix / value_schema_latest / disabled), redpanda.iceberg.delete, redpanda.iceberg.partition.spec, redpanda.iceberg.target.lag.ms, redpanda.iceberg.invalid.record.action (drop / dlq_table). Requires Tiered Storage.
• Server-side Schema ID Validation (reject unregistered schema IDs on destination topics): cluster enable_schema_id_validation (none / redpanda / compat); per-topic redpanda.value.schema.id.validation, redpanda.value.subject.name.strategy, and the key equivalents.
• Tiered Storage (long retention of CDC topics; prerequisite for Iceberg): cluster cloud_storage_enabled; per-topic redpanda.remote.write, redpanda.remote.read, redpanda.remote.delete.
• Cloud Topics (object-storage-native topics): per-topic redpanda.cloud_topic.enabled / redpanda.storage.mode.

Reference Directory

• config-reference.md: Complete field-by-field reference for every microsoft_sql_server_cdc config option, types, defaults, validation rules, and the checkpoint cache internals.
• setup-sqlserver.md: T-SQL walkthrough for enabling SQL Server CDC — sys.sp_cdc_enable_db, sys.sp_cdc_enable_table, SQL Server Agent requirements, capture instance naming, permissions, and Azure SQL specifics.
• pipeline-and-output.md: Full runnable pipelines, metadata shape and routing, snapshot-then-stream lifecycle, checkpointing semantics, and restart behaviour.
• enterprise-features.md: Redpanda Enterprise features for CDC pipelines — Connect license application (flag, env vars, default file path), Iceberg Topics (iceberg_enabled + all redpanda.iceberg.* topic keys and modes), server-side Schema ID Validation (enable_schema_id_validation + per-topic validation/subject-name-strategy keys), Tiered Storage (cloud_storage_enabled, redpanda.remote.*), Cloud Topics (redpanda.cloud_topic.enabled), and license-expiration behavior. Notes which keys require an Enterprise license.