confluentinc/agent-skills/confluent-cloud-cdc-tableflow

Confluent Cloud CDC to Tableflow Pipeline

Build production-ready Change Data Capture pipelines that stream database changes through Confluent Cloud to Iceberg or Delta Lake tables using Debezium, Flink, and Tableflow.

Overview

This skill automates the setup of a complete CDC pipeline:

Database → Debezium CDC Connector → Kafka + Schema Registry → Flink (decode & transform) → Tableflow → Iceberg/Delta Tables

Supported Databases (Fully-Managed Debezium Connectors Only)

• Microsoft SQL Server CDC Source V2
• MySQL CDC Source V2
• PostgreSQL CDC Source V2
• Oracle XStream CDC Source
• DynamoDB CDC Source

Key Components

1. Debezium CDC Source Connector: Captures database changes as events
2. Schema Registry: Manages Avro/JSON/Protobuf schemas (default: JSON_SR)
3. Confluent Cloud Flink: Decodes Debezium envelopes and transforms data
4. Tableflow: Native Confluent Cloud feature that materializes Kafka topics as Iceberg or Delta tables

Critical Architecture Rules

1. NEVER enable Tableflow directly on CDC source topics.

Always use the Flink decode pattern: CDC Source Topic → Flink INSERT → Target Topic (changelog.mode = 'upsert') → Tableflow.

CDC connectors with tombstones.on.delete=true produce null-value Kafka records (tombstones) on DELETE operations. If Tableflow is enabled directly on the CDC source topic, it will use APPEND mode by default and immediately suspend when it encounters a tombstone: "Tableflow will be suspended because we detected a Kafka record with a null value."

The Flink decode layer solves this by interpreting Debezium CDC semantics natively — it translates DELETEs into proper retract/tombstone messages that upsert-mode Tableflow handles correctly.

Do NOT use `after.state.only=true` as a shortcut to bypass the Flink decode step. While it strips the Debezium envelope, tombstone records from DELETEs still break APPEND-mode Tableflow. Additionally, OracleXStreamSource does not support the after.state.only configuration option at all.

2. Tableflow changelog mode is IMMUTABLE after first materialization.

Tableflow caches the changelog mode (APPEND or UPSERT) when it first materializes data. Once set, it cannot be changed — even by altering the Kafka topic's changelog.mode property or by deleting and recreating the Tableflow topic. The S3 table_path is keyed by Kafka topic name, so recreating a Tableflow topic reuses the same S3 path and cached state.

Attempting to change the mode causes: "The changelog mode for this topic has been modified since table materialization began." Flip-flopping the mode further corrupts state with: "Incompatible schema evolution detected."

To change changelog mode, you must delete the Tableflow topic, delete the underlying Kafka topic, and recreate both from scratch. This is why it's critical to create target topics with 'changelog.mode' = 'upsert' from the start.

3. Pipeline cleanup order matters.

When resetting a CDC-to-Tableflow pipeline, delete resources in this order:

1. Tableflow topics (on target topics)
2. Flink INSERT statements
3. Flink target tables (DROP TABLE)
4. Target Kafka topics
5. CDC connectors
6. CDC source Kafka topics (including dbhistory/schema-changes topics)
7. All associated schemas from Schema Registry (both -key and -value subjects)

Never delete CDC source Kafka topics while the connector is still running — the connector cannot recover or re-snapshot and must be fully recreated.

Important Clarifications

• Tableflow is NOT a connector. It is a native topic-level feature enabled via the Tableflow API or Confluent Cloud UI.
• Confluent Cloud Flink auto-discovers CDC tables. You do NOT need to manually create source tables — topics with Schema Registry schemas are automatically available as Flink tables.
• Topics without SR schemas can still be handled — register a JSON schema (partial is fine), use schema inference, or use Flink's raw BYTES with JSON functions. See references/connector-configs.md "Handling Topics Without Schema Registry".
• All SR-backed formats work identically — JSON_SR, AVRO, and PROTOBUF all support Flink auto-discovery and Tableflow. Choose based on throughput needs vs. debuggability.
• Managed connectors use `output.data.format`, not key.converter/value.converter classes.

Workflow Phases

Phase 0: Tool Selection & MCP Server Validation (CRITICAL)

Default: Use Confluent MCP Server. The MCP server is the preferred method for all Confluent Cloud operations. Only fall back to the Confluent CLI (confluent command) and REST APIs if the MCP server is not installed or unavailable.

0.1 Verify MCP Server Availability

Check for mcp__confluent__* tools (list-environments, create-connector, create-flink-statement, create-tableflow-topic, list-schemas, list-topics, consume-messages, search-topics-by-name).

If MCP is not available, fall back to the Confluent CLI (confluent command) and REST APIs for all operations. The CLI fallback should mirror the same workflow phases but use CLI commands instead of MCP tool calls.

CLI Fallback Examples:

# Environment & cluster discovery
confluent environment list
confluent kafka cluster list --environment <env-id>
 
# Connector operations
confluent connect cluster create --config-file connector-config.json --cluster <cluster-id> --environment <env-id>
confluent connect cluster describe <connector-id>
confluent connect cluster list --cluster <cluster-id> --environment <env-id>
 
# Flink operations
confluent flink compute-pool create <pool-name> --cloud <cloud> --region <region> --environment <env-id>
confluent flink statement create <statement-name> --sql "<SQL>" --compute-pool <pool-id> --environment <env-id>
confluent flink statement describe <statement-name> --environment <env-id>
confluent flink statement delete <statement-name> --environment <env-id>
 
# Topic & schema operations
confluent kafka topic list --cluster <cluster-id> --environment <env-id>
confluent schema-registry subject list --environment <env-id>
 
# Tableflow operations
confluent tableflow topic enable <topic-name> --cluster <cluster-id> --environment <env-id> --storage-type MANAGED --table-formats ICEBERG
confluent tableflow topic list --cluster <cluster-id> --environment <env-id>
confluent tableflow topic describe <topic-name> --cluster <cluster-id> --environment <env-id>
confluent tableflow topic disable <topic-name> --cluster <cluster-id> --environment <env-id>

REST API Fallback: If neither MCP nor CLI is available, use the Confluent Cloud REST APIs directly. All calls use HTTP Basic Auth with a Cloud API Key (not a Kafka API key). See references/rest-api.md for endpoint patterns and examples.

0.2 Gather Confluent Cloud Details from the User

Ask the user to provide the following Confluent Cloud details:

Credentials: Generate a cdc-credentials.properties file with placeholders for: Kafka API Key/Secret (cluster-scoped), Database Username/Password. Have the user populate it in their editor and add it to .gitignore. If the user prefers Claude not read the file, fall back to CLI: generate connector-config.json with placeholders, user fills it in, then confluent connect cluster create --config-file connector-config.json.

0.3 Verify MCP Cluster Targeting

Quick verification:

1. Run mcp__confluent__list-topics to confirm the MCP server is connected to the expected cluster
2. Run mcp__confluent__list-schemas to verify Schema Registry is accessible

Schema Registry is shared at the environment level across all clusters.

Phase 1: Discovery & Validation

1.1 Check Existing Setup

Use MCP to check what already exists:

mcp__confluent__list-connectors(environmentId, clusterId)  →  Existing CDC connectors
mcp__confluent__list-flink-statements(environmentId, computePoolId)  →  Existing Flink jobs
mcp__confluent__list-tableflow-topics(environmentId, clusterId)  →  Existing Tableflow topics

Ask the user:

• "Do you have any CDC connectors already running?"
• "Do you have a Flink compute pool you'd like to use, or should I create one?"
• "Is your database already configured for CDC?"

1.2 Validate Topic Prefix Uniqueness

Before proceeding, validate that the chosen topic.prefix won't collide with existing topics:

mcp__confluent__search-topics-by-name(topicName: "<proposed-prefix>", environmentId, clusterId)

Or via CLI:

confluent kafka topic list --cluster <cluster-id> --environment <env-id> | grep "^<proposed-prefix>"

If any existing topics share the proposed prefix, warn the user and recommend a unique prefix. A prefix collision silently merges CDC data with unrelated topics, which can corrupt both pipelines.

1.3 Check Schema Registry Compatibility

Default BACKWARD compatibility can halt CDC connectors when database columns are dropped. Set FULL_TRANSITIVE for CDC subjects after the connector creates them:

confluent schema-registry config update --subject "<topic-prefix>.<schema>.<table>-value" --compatibility FULL_TRANSITIVE --environment <env-id>

1.4 Gather Required Information

Database Configuration:

• Database type (SQL Server, MySQL, PostgreSQL, Oracle, or DynamoDB)
• Connection details (hostname, port, database name)
• Credentials (populated by the user in the credentials file)
• Specific tables to capture (format: schema.table)

Schema Format: Ask the user: JSON_SR (default, human-readable), AVRO (smaller payloads, high-throughput), or PROTOBUF (strongly typed). All work identically with Flink auto-discovery and Tableflow. Never use plain `JSON` — it breaks both. See references/connector-configs.md for detailed comparison.

Existing Topics Without SR: See references/connector-configs.md "Handling Topics Without Schema Registry" for options (register JSON schema, schema inference, or Flink raw BYTES).

Tableflow Destination:

• Target format: Iceberg or Delta Lake
• Storage: Managed (recommended, Confluent manages S3) or BYOB (user's own S3 bucket, requires Provider Integration ID)

Naming Convention:

• Default: cdc-pipeline-skill-{database-type}-{YYYYMMDD}
• Example: cdc-pipeline-skill-postgres-20260323

1.5 Validate Database Prerequisites

Each database requires specific CDC setup. Read references/database-prerequisites.md for details:

• PostgreSQL: WAL level = logical, replication slots, publication
• MySQL: binlog format = ROW, GTID mode
• SQL Server: CDC enabled on database and tables, SQL Server Agent running
• Oracle XStream: GoldenGate replication enabled (`enable_goldengate_replication=TRUE`), ARCHIVELOG mode, supplemental logging, XStream admin user with `DBMS_XSTREAM_AUTH` privileges, XStream outbound server created via `DBMS_XSTREAM_ADM.CREATE_OUTBOUND`, connector user with XStream connect privilege. Full prereqs: https://docs.confluent.io/cloud/current/connectors/cc-oracle-xstream-cdc-source/prereqs-validation.md
• DynamoDB: Streams enabled with NEW_AND_OLD_IMAGES

If the database isn't properly configured, guide the user through setup before proceeding.

Oracle XStream important limitations:

• Only supports non-CDB architecture on Amazon RDS for Oracle
• Does NOT support Oracle Autonomous Databases or Oracle Standby (Data Guard)
• Does NOT support Downstream Capture
• after.state.only is NOT supported by OracleXStreamSource
• Requires a valid Confluent license for XStream Out

Phase 2: Planning