confluent-cloud-cdc-tableflow

Files (7)

• SKILL.md
• evals.json · evals/
• connector-configs.md · references/
• database-prerequisites.md · references/
• flink-sql-patterns.md · references/
• rest-api.md · references/
• troubleshooting.md · references/

SKILL.md (first lines)

# 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:**
```bash
# 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:

| Detail | Example | Used For |
|---|---|---|
| Environment ID | `env-0ypxv6` | `environmentId` in all MCP calls |
| Kafka Cluster ID | `lkc-qo5k36` | `clusterId` in all MCP calls |
| Flink Compute Pool ID | `lfcp-3v39xw` | `computePoolId` in Flink statements |
| Flink Catalog Name | `my_environment` | `catalogName` in Flink statements |
| Flink Database Name | `cluster_0` | `databaseName` in Flink statements |

**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`.

From confluentinc/agent-skills, Apache-2.0.

More from confluentinc/agent-skills