confluent-cloud-cdc-tableflow

Confluent Cloud CDC から Tableflow へのパイプライン

Debezium、Flink、Tableflow を使用して、データベースの変更を Confluent Cloud 経由で Iceberg または Delta Lake のテーブルにストリーミングする、本番環境に対応した変更データキャプチャ（CDC）パイプラインを構築します。

概要

このスキルでは、完全な CDC パイプラインのセットアップを自動化します。

データベース → Debezium CDC コネクター → Kafka + Schema Registry → Flink（デコードと変換） → Tableflow → Iceberg／Delta テーブル

対応データベース（完全マネージド型 Debezium コネクターのみ）

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

主なコンポーネント

1. Debezium CDC Source Connector：データベースの変更をイベントとしてキャプチャします
2. Schema Registry：Avro／JSON／Protobuf スキーマを管理します（デフォルト：JSON_SR）
3. Confluent Cloud Flink：Debezium のエンベロープをデコードし、データを変換します
4. Tableflow：Kafka トピックを Iceberg または Delta テーブルとして具現化する、Confluent Cloud のネイティブ機能です

重要なアーキテクチャルール

1. Tableflow を CDC ソーストピックで直接有効にしないでください。

必ず Flink のデコードパターンを使用してください：CDC ソーストピック → Flink INSERT → ターゲットトピック（changelog.mode = 'upsert'） → Tableflow。

tombstones.on.delete=true を設定した CDC コネクターは、DELETE 操作時に値が null の Kafka レコード（トゥームストーン）を生成します。Tableflow を CDC ソーストピックで直接有効にすると、デフォルトでは APPEND モードが使用され、トゥームストーンに遭遇した時点で直ちに一時停止します。「Kafka レコードの値が null であることを検出したため、Tableflow は一時停止されます。」

Flink のデコードレイヤーでは、Debezium の CDC セマンティクスがネイティブに解釈されます。DELETE を適切なリトラクト／トゥームストーンメッセージに変換するため、UPSERT モードの Tableflow で正しく処理できます。

Flink のデコード手順を省略するための近道として、after.state.only=true を使用しないでください。Debezium のエンベロープは取り除かれますが、DELETE によるトゥームストーンレコードは APPEND モードの Tableflow を引き続き停止させます。また、OracleXStreamSource は after.state.only 設定オプションをサポートしていません。

2. Tableflow の変更履歴モードは、最初の具現化後に変更できません。

Tableflow は、最初にデータを具現化したときに変更履歴モード（APPEND または UPSERT）をキャッシュします。一度設定されると、Kafka トピックの changelog.mode プロパティを変更したり、Tableflow トピックを削除して再作成したりしても変更できません。S3 の table_path は Kafka トピック名をキーとしているため、Tableflow トピックを再作成すると、同じ S3 パスとキャッシュ済みの状態が再利用されます。

モードを変更しようとすると、次のエラーが発生します。「テーブルの具現化が始まってから、このトピックの変更履歴モードが変更されました。」 モードをさらに切り替えると、状態が破損し、次のエラーが発生します。「互換性のないスキーマ進化が検出されました。」

変更履歴モードを変更するには、Tableflow トピックと基盤となる Kafka トピックを削除してから、両方を最初から再作成する必要があります。そのため、ターゲットトピックは最初から 'changelog.mode' = 'upsert' を指定して作成することが重要です。

3. パイプラインのクリーンアップ順序は重要です。

CDC から Tableflow へのパイプラインをリセットするときは、次の順序でリソースを削除してください。

1. Tableflow トピック（ターゲットトピック上）
2. Flink INSERT ステートメント
3. Flink ターゲットテーブル（DROP TABLE）
4. ターゲット Kafka トピック
5. CDC コネクター
6. CDC ソース Kafka トピック（dbhistory／schema-changes トピックを含む）
7. Schema Registry の関連スキーマすべて（-key と -value の両方のサブジェクト）

コネクターの実行中に CDC ソース Kafka トピックを削除しないでください。コネクターは復旧も再スナップショットもできず、完全に再作成する必要があります。

重要な補足

• Tableflow はコネクターではありません。 Tableflow API または Confluent Cloud UI から有効にする、トピックレベルのネイティブ機能です。
• Confluent Cloud Flink は CDC テーブルを自動検出します。 ソーステーブルを手動で作成する必要はありません。Schema Registry のスキーマがあるトピックは、Flink テーブルとして自動的に利用可能になります。
• SR スキーマのないトピックも処理できます。JSON スキーマを登録する（一部のみでも可）、スキーマ推論を使用する、または Flink の生の BYTES と JSON 関数を使用します。references/connector-configs.md の「Handling Topics Without Schema Registry」を参照してください。
• SR を使用する形式はすべて同じように動作します。 JSON_SR、AVRO、PROTOBUF はすべて、Flink の自動検出と Tableflow に対応しています。スループットのニーズとデバッグのしやすさを考慮して選択してください。
• マネージドコネクターでは key.converter／value.converter クラスではなく、output.data.format を使用します。

ワークフローのフェーズ

フェーズ 0：ツールの選択と MCP サーバーの検証（重要）

デフォルト：Confluent MCP Server を使用します。 Confluent Cloud のすべての操作では、MCP サーバーを優先して使用します。MCP サーバーがインストールされていない、または利用できない場合に限り、Confluent CLI（confluent コマンド）と REST API に切り替えます。

0.1 MCP サーバーの利用可否を確認する

mcp__confluent__* ツール（list-environments、create-connector、create-flink-statement、create-tableflow-topic、list-schemas、list-topics、consume-messages、search-topics-by-name）を確認します。

MCP を利用できない場合は、すべての操作で Confluent CLI（confluent コマンド）と REST API に切り替えます。CLI による代替手段では、MCP ツール呼び出しの代わりに CLI コマンドを使用し、同じワークフローのフェーズに沿って進めてください。

CLI による代替例：

# 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 による代替： MCP も CLI も利用できない場合は、Confluent Cloud REST API を直接使用します。すべての呼び出しには、Cloud API Key（Kafka API キーではありません）を使った HTTP Basic 認証を使用します。エンドポイントのパターンと例については、references/rest-api.md を参照してください。

0.2 ユーザーから Confluent Cloud の詳細情報を収集する

次の Confluent Cloud の詳細情報をユーザーに尋ねます。

詳細情報	例	用途
Environment ID	env-0ypxv6	すべての MCP 呼び出しの environmentId
Kafka Cluster ID	lkc-qo5k36	すべての MCP 呼び出しの clusterId
Flink Compute Pool ID	lfcp-3v39xw	Flink ステートメントの computePoolId
Flink Catalog Name	my_environment	Flink ステートメントの catalogName
Flink Database Name	cluster_0	Flink ステートメントの databaseName

認証情報： Kafka API Key／Secret（クラスター単位）と Database Username／Password のプレースホルダーを含む cdc-credentials.properties ファイルを作成します。ユーザーにエディターで値を入力してもらい、.gitignore に追加します。Claude にファイルを読み取らせたくない場合は、CLI に切り替えます。プレースホルダーを含む connector-config.json を作成し、ユーザーに値を入力してもらった後、confluent connect cluster create --config-file connector-config.json を実行します。

0.3 MCP の接続先クラスターを検証する

簡易検証：

1. mcp__confluent__list-topics を実行し、MCP サーバーが想定したクラスターに接続されていることを確認します
2. mcp__confluent__list-schemas を実行し、Schema Registry にアクセスできることを確認します

Schema Registry は環境レベルで共有され、すべてのクラスターから利用できます。

フェーズ 1：検出と検証

1.1 既存の設定を確認する

MCP を使って、すでに存在するものを確認します。

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

ユーザーに尋ねます。

• 「すでに稼働している CDC コネクターはありますか？」
• 「使用したい Flink コンピュートプールはありますか？それともこちらで作成しますか？」
• 「データベースはすでに CDC 用に設定されていますか？」

1.2 トピックプレフィックスの一意性を検証する

先に進む前に、選択した topic.prefix が既存のトピックと衝突しないことを確認します。

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

または CLI を使用します。

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

提案したプレフィックスを共有する既存のトピックがある場合は、ユーザーに警告し、一意のプレフィックスを推奨します。プレフィックスが衝突すると、CDC データと無関係なトピックが気付かれないまま統合され、両方のパイプラインが破損する可能性があります。

1.3 Schema Registry との互換性を確認する

デフォルトの BACKWARD 互換性では、データベースの列が削除されたときに CDC コネクターが停止することがあります。コネクターが作成した後、CDC のサブジェクトには FULL_TRANSITIVE を設定してください。

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

1.4 必要な情報を収集する

データベースの設定：

• データベースの種類（SQL Server、MySQL、PostgreSQL、Oracle、または DynamoDB）
• 接続情報（ホスト名、ポート、データベース名）
• 認証情報（ユーザーが認証情報ファイルに入力）
• キャプチャする特定のテーブル（形式：schema.table）

スキーマ形式： ユーザーに JSON_SR（デフォルト、人間が読みやすい）、AVRO（ペイロードが小さく、高スループット）、または PROTOBUF（厳密に型付けされる）のいずれかを確認してください。すべて Flink の自動検出および Tableflow と同じように動作します。プレーンな JSON は絶対に使用しないでください。どちらも動作しなくなります。詳しい比較については、references/connector-configs.md を参照してください。

SR を使用しない既存のトピック： 選択肢（JSON スキーマの登録、スキーマ推論、Flink の raw BYTES）については、references/connector-configs.md の「Handling Topics Without Schema Registry」を参照してください。

Tableflow の宛先：

• 対象形式：Iceberg または Delta Lake
• ストレージ：マネージド（推奨。Confluent が S3 を管理）または BYOB（ユーザー所有の S3 バケット。Provider Integration ID が必要）

命名規則：

• デフォルト：cdc-pipeline-skill-{database-type}-{YYYYMMDD}
• 例：cdc-pipeline-skill-postgres-20260323

1.5 データベースの前提条件を確認する

各データベースでは、特定の CDC 設定が必要です。詳細については references/database-prerequisites.md を参照してください。

• PostgreSQL：WAL レベル = logical、レプリケーションスロット、publication
• MySQL：binlog 形式 = ROW、GTID モード
• SQL Server：データベースとテーブルで CDC が有効、SQL Server Agent が実行中
• Oracle XStream：GoldenGate レプリケーションが有効（enable_goldengate_replication=TRUE）、ARCHIVELOG モード、補足ログ、DBMS_XSTREAM_AUTH 権限を持つ XStream 管理者ユーザー、DBMS_XSTREAM_ADM.CREATE_OUTBOUND で作成した XStream outbound server、XStream 接続権限を持つコネクタユーザー。前提条件の詳細：https://docs.confluent.io/cloud/current/connectors/cc-oracle-xstream-cdc-source/prereqs-validation.md
• DynamoDB：Streams が NEW_AND_OLD_IMAGES で有効

データベースが適切に設定されていない場合は、先に設定を行うようユーザーを案内してください。

Oracle XStream の重要な制限事項：

• Amazon RDS for Oracle の非 CDB アーキテクチャのみをサポート
• Oracle Autonomous Databases または Oracle Standby（Data Guard）はサポートしません
• Downstream Capture はサポートしません
• after.state.only は OracleXStreamSource でサポートされません
• XStream Out には有効な Confluent ライセンスが必要です

フェーズ 2：計画

設定計画全体を作成し、承認を得るためにユーザーに提示します。

2.1 コネクターの設定

データベースの種類に基づき、references/connector-configs.md の該当するテンプレートを使用してコネクターの設定を作成します。テンプレートには、必須フィールド（name、connector.class、topic.prefix、kafka.api.key、output.data.format、decimal.handling.mode など）と、データベース固有の設定がすべて含まれています。

ユーザーの希望に基づいてスキーマ形式を設定します（デフォルトは JSON_SR）：

{
  "output.data.format": "JSON_SR",
  "output.key.format": "JSON_SR"
}

ユーザーが別の形式を希望した場合は、JSON_SR を AVRO または PROTOBUF に置き換えてください。キーと値の形式は一致させてください。形式の選択にかかわらず、その他のコネクター設定はすべて同じです。

トピックの命名パターン： {topic.prefix}.{schema}.{table} topic.prefix = "postgres-cdc" の場合の例：postgres-cdc.public.customers

2.2 Flink SQL ステートメント

Confluent Cloud Flink では、CDC ソーステーブルは Kafka トピックから自動検出されます。必要な作業は以下だけです。

注： 以下の例では、説明用のカラム名を持つ customers テーブルを使用しています。CDC トピックから検出されたスキーマに基づき、ユーザーの実際のテーブル名、カラム、型に置き換えてください。

1. ターゲットテーブルを作成する（Tableflow へのプレーンな JSON_SR 出力の場合）：

CREATE TABLE `target_customers` (
  `id` INT NOT NULL,
  `name` STRING,
  `email` STRING,
  `created_at` TIMESTAMP_LTZ(3),
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'changelog.mode' = 'upsert'
);

2. INSERT ステートメントを作成する（デコードと変換を行う継続ジョブ）：

INSERT INTO `target_customers`
SELECT
  `id`,
  `name`,
  `email`,
  TO_TIMESTAMP_LTZ(`created_at` / 1000, 3)
FROM `postgres-cdc.public.customers`;

Cloud Flink での重要な相違点：

• CREATE TABLE では 'connector'、'value.format'、'properties.bootstrap.servers'、Schema Registry の URL を指定しないでください。Cloud Flink がこれらすべてを自動的に処理します。
• CDC トピックのソーステーブルを作成しないでください。自動検出されます。
• after.* フィールドを参照したり、op でフィルタリングしたりしないでください。Flink は CDC の変更ログのセマンティクスをネイティブに解釈します。
• Debezium のタイムスタンプには TIMESTAMP(3) ではなく TIMESTAMP_LTZ(3) を使用してください。

DynamoDB CDC は、Flink における SQL CDC とは異なります。 自動検出されるテーブルには、カラム id（キー）と val（op、before.document、after.document を含む CDC エンベロープを格納した複合 ROW 型）があります。Flink は SQL Debezium コネクターの場合のように、このエンベロープを自動デコードしません。フィールドを手動で抽出する必要があります。

CREATE TABLE `target_dynamodb` (
  `id` STRING NOT NULL,
  `document` STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH ('changelog.mode' = 'upsert');

INSERT INTO `target_dynamodb`
SELECT `id`, `val`.`after`.`document`
FROM `dynamodb-cdc-source-topic`;

document フィールドは、DynamoDB ネイティブの型形式で DynamoDB 項目を表す JSON 文字列です（例：{"name":{"S":"Alice"},"age":{"N":"30"}}）。

Debezium の型変換： 型マッピングの完全な表については、references/flink-sql-patterns.md を参照してください。主な変換は、MicroTimestamp には TO_TIMESTAMP_LTZ(col / 1000, 3)、Timestamp には TO_TIMESTAMP_LTZ(col, 3) を使用し、コネクターで decimal.handling.mode=string が設定されていることを確認してください（BYTES のデフォルト値は Flink では使用できません）。

2.3 Tableflow の設定

Tableflow はトピックレベルのネイティブ機能であり、コネクターではありません。トピックごとに有効化します。

ストレージオプション：

• Managed（推奨）：Confluent が S3 ストレージを管理します。認証情報は不要です。
• BYOB（Bring Your Own Bucket）：ユーザーが S3 バケットを指定します。Confluent Cloud で Provider Integration ID を設定する必要があります（Settings → Provider Integrations）。

テーブル形式： Iceberg（推奨）または Delta Lake

2.4 計画を提示する

ユーザーに以下を提示します。

1. コネクターの設定（機密フィールドをマスク）
2. 使用する Flink コンピュートプール
3. Flink SQL ステートメント（ターゲットテーブル＋INSERT）
4. Tableflow の設定（ストレージタイプ、形式）
5. 予想されるトピック名

続行する前に、明示的な確認を待ちます。

フェーズ 3：実行

MCP ツールを使用してステップごとに実行し、各コンポーネントの後にステータスを確認します。

3.1 CDC ソースコネクターを作成する

references/connector-configs.md にある、ユーザーのデータベースタイプ用テンプレートを使用してコネクター設定を作成します。各テンプレートには name フィールドを含む、必要なフィールドがすべて含まれています。

MCP の使用：

mcp__confluent__create-connector(
  connectorName: "<connector-name>",
  environmentId: "<env-id>",
  clusterId: "<cluster-id>",
  connectorConfig: { <config from references/connector-configs.md> }
)

確認： マネージドコネクターのプロビジョニングには2～5 分かかります。mcp__confluent__read-connector をポーリングします。tasks: [] はプロビジョニング中、tasks: [{...}] は準備完了を意味します。次に、mcp__confluent__list-schemas(subjectPrefix: "postgres-cdc") でスキーマを確認します。タスクが割り当てられている状態で 5 分経過してもスキーマがない場合は、データベースへの接続を確認してください。コネクターのエラーログは Confluent Cloud の UI を使用して確認してください（MCP では取得できません）。

3.2 Flink コンピュートプールを作成する（必要な場合）

既存の Flink コンピュートプールがない場合は、SQL を実行する前に作成します。

confluent flink compute-pool create <pool-name> --cloud <cloud-provider> --region <region> --environment <env-id>

Kafka クラスターと同じクラウドプロバイダーおよびリージョンを使用します。プールのステータスが RUNNING になるまで待ってから続行してください。

3.3 Flink SQL を実行する

手順 1：CDC テーブルが自動検出されていることを確認します：

mcp__confluent__create-flink-statement(
  statementName: "show-tables-check",
  statement: "SHOW TABLES;",
  environmentId: "<env-id>",
  computePoolId: "<pool-id>",
  catalogName: "<environment-display-name>",
  databaseName: "<cluster-display-name>"
)

次に、結果を読み取ります：

mcp__confluent__read-flink-statement(statementName: "show-tables-check", environmentId: "<env-id>")

CDC トピックテーブル（例：postgres-cdc.public.customers）を探します。存在しない場合は、コネクターがまだデータを生成していません。待ってから再試行してください。

手順 2：ターゲットテーブルを作成します：

mcp__confluent__create-flink-statement(
  statementName: "cdc-create-target-customers",
  statement: "CREATE TABLE `target_customers` (...) WITH ('changelog.mode' = 'upsert');",
  environmentId, computePoolId, catalogName, databaseName
)

手順 3：INSERT ジョブを作成します：

mcp__confluent__create-flink-statement(
  statementName: "cdc-decode-customers",
  statement: "INSERT INTO `target_customers` SELECT ... FROM `postgres-cdc.public.customers`;",
  environmentId, computePoolId, catalogName, databaseName
)

INSERT によって、継続的に実行される Flink ジョブが作成されます。FAILED ではなく RUNNING に遷移することを確認します：

mcp__confluent__read-flink-statement(statementName: "cdc-decode-customers", environmentId)

よくある INSERT の失敗：

• "Table does not exist" → CDC ソーステーブルがまだ自動検出されていません。コネクターを待ってください
• "Incompatible types for sink column" → 型が一致していません。上記の Debezium の型マッピングを確認してください
• "Unsupported format" → CREATE TABLE から明示的なフォーマットプロパティを削除してください

アドバイザリー警告（無視してかまいません）：

• "Primary key does not match upsert key" — CDC デコードパターンでは想定される警告です
• "Highly state-intensive operators without TTL" — アドバイザリー警告です。本番環境で必要に応じて TTL を設定してください

3.4 Tableflow を有効にする

MCP を使用する場合：

mcp__confluent__create-tableflow-topic(
  tableflowTopicConfig: {
    "display_name": "target_customers",
    "storage": { "kind": "Managed", "bucket_name": "managed", "provider_integration_id": "managed" },
    "table_formats": ["ICEBERG"],
    "config": { "record_failure_strategy": "SUSPEND", "retention_ms": "6048000000" }
  }
)

既知の制限事項： MCP の create-tableflow-topic ツールは、environmentId または clusterId パラメーターを受け付けません。MCP サーバーで設定されているクラスターがデフォルトで使用されます。MCP サーバーがターゲットトピックのあるクラスターとは別のクラスターを参照している場合、「topic not found」エラーで失敗します。回避策として CLI または UI を使用してください。

CLI を使用する場合：

# Managed storage (Confluent manages S3)
confluent tableflow topic enable target_customers \
  --cluster <cluster-id> \
  --environment <env-id> \
  --storage-type MANAGED \
  --table-formats ICEBERG

# BYOB / BYOS (user's own S3 bucket)
confluent tableflow topic enable target_customers \
  --cluster <cluster-id> \
  --environment <env-id> \
  --storage-type BYOS \
  --provider-integration <provider-integration-id> \
  --bucket-name <bucket-name> \
  --table-formats ICEBERG

# Azure Data Lake Storage Gen2
confluent tableflow topic enable target_customers \
  --cluster <cluster-id> \
  --environment <env-id> \
  --storage-type AzureDataLakeStorageGen2 \
  --provider-integration <provider-integration-id> \
  --storage-account-name <account-name> \
  --container-name <container-name> \
  --table-formats DELTA

Iceberg の代わりに Delta Lake を使用する場合は、--table-formats DELTA を使用してください。

Tableflow が有効になっていることを確認します：

mcp__confluent__list-tableflow-topics(environmentId, clusterId)

または、CLI を使用します：

confluent tableflow topic describe target_customers --cluster <cluster-id> --environment <env-id>
confluent tableflow topic list --cluster <cluster-id> --environment <env-id>

ステータスは PENDING から ACTIVE に遷移します。

フェーズ 4：検証とトラブルシューティング

4.1 エンドツーエンドのパイプラインを検証する

大規模テーブルのスナップショット： 大規模テーブルに対して snapshot.mode: initial を指定してコネクターを作成した場合、検証に数時間かかることがあります。スナップショットが実行中なのか、パイプラインが壊れているのかを見分けるには：

1. コネクターにタスクが割り当てられていることを確認します（read-connector → tasks が空でない）
2. スキーマが登録されていることを確認します（list-schemas → キー／値スキーマが存在する）。これはスナップショットによるデータ生成が開始されていることを意味します
3. Confluent Cloud UI でソーストピックのメッセージ数を監視します。継続的にメッセージが流れていれば、処理は進行しています

初期検証に snapshot.mode: schema_only を使用した場合は、ソースデータベースにテスト行を挿入して CDC イベントを発生させ、パイプライン全体を検証してください。スナップショットの詳細なトラブルシューティングについては、references/troubleshooting.md を参照してください。

各コンポーネントを確認します：

確認項目	MCP ツール	確認する内容
コネクターの実行状態	read-connector	tasks 配列が空でない
スキーマの登録状態	list-schemas(subjectPrefix)	CDC トピックのキーと値のスキーマ
Flink 内の CDC テーブル	create-flink-statement("SHOW TABLES")	CDC トピックがテーブルとして表示される
Flink ジョブの実行状態	read-flink-statement	レスポンスにエラーがない
ターゲットトピックのデータ	consume-messages(topicNames)	メッセージが表示される（注：コンシューマーは最新のオフセットから開始します）
Tableflow の有効化状態	list-tableflow-topics	ステータスが PENDING または ACTIVE

ターゲットトピックからメッセージを消費して、デコードされたデータを検証します：

mcp__confluent__consume-messages(
  topicNames: ["target_customers"],
  value: { "useSchemaRegistry": true },
  key: { "useSchemaRegistry": true },
  maxMessages: 5,
  timeoutMs: 15000
)

注：コンシューマーは最新のオフセットから開始します。初期スナップショットがすでに完了している場合、新たなデータベース変更が発生するまでメッセージが 0 件と表示されることがあります。

ソースデータベースに行を挿入してリアルタイム CDC をテストします（テーブル名と列をユーザーの実際のスキーマに合わせて変更してください）：

INSERT INTO public.customers (name, email, created_at)
VALUES ('Test User', '[email protected]', NOW());

4.2 トラブルシューティング

詳しいトラブルシューティングについては、references/troubleshooting.mdを参照してください。

クイックリファレンス — パイプラインを停止させる問題：

症状	考えられる原因	対処方法
コネクタータスクが空のまま	プロビジョニング中	2～5分待ってから、再試行してください
5分経ってもスキーマがない	DB 接続または認証情報	ホスト、ポート、ユーザー、パスワードを確認し、DB CDC 設定を検証してください
SHOW TABLES に CDC テーブルが表示されない	コネクターがまだデータを生成していない	まずスキーマが存在することを確認してから、しばらく待ってください
INSERT：「Incompatible types」	Debezium の型の不一致	TIMESTAMP_LTZ(3) + TO_TIMESTAMP_LTZ を使用してください。references/flink-sql-patterns.mdを参照してください
Tableflow：「topic not found」	MCP クラスターの不一致	CLI で confluent tableflow topic enable を実行するか、Confluent Cloud UI を使用してください
トピックは存在するが、SHOW TABLES に表示されない	SR にスキーマがない	SR に JSON スキーマを登録するか、スキーマ推論を使用してください。references/connector-configs.mdを参照してください

Phase 5：ドキュメント

セットアップが正常に完了したら、ユーザーに次の情報を提供してください。

1. パイプライン概要表：すべてのコンポーネント名、ID、ステータス
2. トピック名：ソース CDC トピックとターゲットトピック（使用したスキーマ形式を含む）
3. モニタリング：Confluent Cloud UI でコネクター、Flink ジョブ、Tableflow のステータスを確認
4. テストコマンド：リアルタイム CDC を検証する SQL INSERT

参考資料

• データベースの前提条件：references/database-prerequisites.md
• コネクターの設定：references/connector-configs.md
• Flink SQL パターン：references/flink-sql-patterns.md
• トラブルシューティングガイド：references/troubleshooting.md
• REST API リファレンス：references/rest-api.md
• Confluent Cloud Flink ドキュメント：https://docs.confluent.io/cloud/current/flink/overview.md
• Tableflow ドキュメント：https://docs.confluent.io/cloud/current/topics/tableflow/overview.md
• Debezium CDC ドキュメント：https://debezium.io/documentation/
• Confluent MCP サーバー：https://github.com/confluentinc/mcp-confluent