JA
FrançaisPortuguêsEspañolDeutschTiếng ViệtEnglish日本語한국어简体中文
ホーム / アーキテクチャ・設計 / Kafka

Change Data CaptureとDebeziumとKafka

Cris The Dev · 2026/10/12 · 8 分で読めます · 312

1. 概要

あるサービスが別のデータベースのデータが変わったことを知る必要がある場合、よくある方法はデータベースに書き込んだ後、メッセージをKafkaに送信することです。この方法はデュアルライトと呼ばれ、データの不整合を引き起こしやすいです。Change Data Capture (CDC)はデータベースのトランザクションログを直接読むことでこの問題を解決します。

このチュートリアルでは、Docker Composeを使ってPostgreSQL、Kafka、Debeziumをセットアップし、connectorを登録します。その後、ordersテーブルのINSERT、UPDATE、DELETEイベントを処理するSpring Bootコンシューマーを作成します。

2. CDCが必要な理由

次の一般的なデュアルライトコード例を見てください：

@Transactional
public void placeOrder(Order order) {
    orderRepository.save(order);
    kafkaTemplate.send("orders", order.getId().toString(), toJson(order));
}

このコードには2つの弱点があります：

  • Kafkaがsave成功後にエラーになった場合、データベースには注文がありますが、イベントは送信されません。
  • sendは成功したがトランザクションがロールバックした場合、コンシューマーは存在しない注文のイベントを受け取ります。

CDCはこのフローを逆にします。アプリケーションはデータベースに書き込むだけです。別のプロセスがトランザクションログ（PostgreSQLの場合はWAL、MySQLの場合はbinlog）を読み、コミットされた各変更に対してイベントを発行します。ログにはコミットされた変更のみが含まれているため、イベントは常に実データと一致します。

Debeziumはkafka Connect上で動作するCDC connectorのセットです。データフローは次のようになります：

アプリケーション ──書き込み──> PostgreSQL ──WAL──> Debezium (Kafka Connect) ──> Kafkaトピック ──> コンシューマー

3. 環境構築

3.1. Docker Compose

3つのコンテナが必要です：logical replicationを有効にしたPostgreSQL、KRaftモードで動作するKafka、Debeziumプラグインが含まれているKafka Connectです。

services:
  postgres:
    image: postgres:16
    command: ["postgres", "-c", "wal_level=logical"]
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: shop
    ports:
      - "5432:5432"

  kafka:
    image: apache/kafka:3.8.0
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:29092,CONTROLLER://:9093,EXTERNAL://:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,EXTERNAL://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1

  connect:
    image: quay.io/debezium/connect:2.7
    depends_on: [kafka, postgres]
    ports:
      - "8083:8083"
    environment:
      BOOTSTRAP_SERVERS: kafka:29092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: connect_configs
      OFFSET_STORAGE_TOPIC: connect_offsets
      STATUS_STORAGE_TOPIC: connect_statuses

このファイルで注意すべき3つのポイントがあります：

  • wal_level=logicalは必須です。このパラメータがないと、Debeziumはレプリケーションスロットを作成できません。
  • Kafkaには2つのリスナーがあります。コンテナはkafka:29092を使用し、マシンで実行されているSpring Bootアプリケーションはlocalhost:9092を使用します。
  • Kafka Connectは設定、オフセット、ステータスを3つの別々のトピックに保存します。これにより、コネクタが再起動してもWAL内の前の位置から読み続けることができます。

起動し、テーブルを作成します：

docker compose up -d
docker compose exec postgres psql -U postgres -d shop -c "
  CREATE TABLE orders (
    id BIGSERIAL PRIMARY KEY,
    customer_email VARCHAR(255) NOT NULL,
    status VARCHAR(32) NOT NULL,
    total NUMERIC(10,2) NOT NULL
  );
  ALTER TABLE orders REPLICA IDENTITY FULL;"

REPLICA IDENTITY FULLはPostgreSQLにすべての古い行をWALに書き込むよう指示します。このコマンドがない場合、UPDATEイベントのbeforeフィールドはnullになり、DELETEイベントには主キーのみが含まれます。

3.2. Connectorの登録

Kafka Connectはポート8083でREST APIを提供します。PostgreSQL connectorを次のように登録します：

curl -X POST -H "Content-Type: application/json" localhost:8083/connectors -d '{
  "name": "shop-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "postgres",
    "database.password": "postgres",
    "database.dbname": "shop",
    "topic.prefix": "shop",
    "table.include.list": "public.orders",
    "slot.name": "shop_slot",
    "publication.autocreate.mode": "filtered",
    "decimal.handling.mode": "string",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
  }
}'

重要なプロパティの意味：

  • plugin.name: pgoutputはPostgreSQL 10+に組み込まれているデコードプラグインを使用するため、追加の拡張機能をインストールする必要がありません。
  • topic.prefixはスキーマとテーブル名と一緒にトピック名を生成します。ここではトピックはshop.public.ordersです。
  • table.include.listはキャプチャされるテーブルを制限します。
  • decimal.handling.mode: stringにより、NUMERICカラムが文字列"149.90"として送信されます。デフォルトでは、Debeziumはこの値をbase64バイトにエンコードし、読みにくくなります。
  • schemas.enable: falseは各JSONメッセージに付属するschema部分を削除し、ペイロードをより簡潔にします。

コネクタのステータスを確認します：

curl -s localhost:8083/connectors/shop-connector/status
{"name":"shop-connector","connector":{"state":"RUNNING","worker_id":"172.18.0.4:8083"},
 "tasks":[{"id":0,"state":"RUNNING","worker_id":"172.18.0.4:8083"}],"type":"source"}

4. Change Eventの構造

注文を追加して更新します：

docker compose exec postgres psql -U postgres -d shop -c "
  INSERT INTO orders (customer_email, status, total) VALUES ('an@example.com', 'NEW', 149.90);
  UPDATE orders SET status = 'PAID' WHERE id = 1;"

その後、Kafkaイメージに含まれているコンソールコンシューマーを使用してトピックを読みます：

docker compose exec kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 --topic shop.public.orders --from-beginning

イベントUPDATEは次のようになります（sourceフィールドは短縮されています）：

{
  "before": {"id":1,"customer_email":"an@example.com","status":"NEW","total":"149.90"},
  "after":  {"id":1,"customer_email":"an@example.com","status":"PAID","total":"149.90"},
  "source": {"connector":"postgresql","db":"shop","table":"orders","lsn":24023128, ...},
  "op": "u",
  "ts_ms": 1760170000123
}

opフィールドは変更のタイプを示します：

op 意味 before after
c INSERT null 新しい行
u UPDATE 古い行 新しい行
d DELETE 古い行 null
r 初期スナップショット内での読み取り null 現在の行

行を削除する場合、Debeziumは追加のtombstone（同じキーを持つメッセージですが、値がnull）を送信します。Tombstoneにより、Kafkaのログコンパクションそのキーを完全に削除できます。コンシューマーはnullの値を処理する必要があります。

5. Spring Bootでイベントを消費する

5.1. 依存関係と設定

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-json</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

上記の依存関係のバージョンはspring-boot-starter-parent 3.xによって管理されます。

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: order-sync
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

ここでコンシューマーはString形式の値を受け取り、自分で解析します。この方法により、JsonDeserializerを使用するよりもtombstoneとJSONエラーをより簡単に制御できます。

5.2. モデルとパーサー

@JsonIgnoreProperties(ignoreUnknown = true)
public record OrderRow(
        Long id,
        @JsonProperty("customer_email") String customerEmail,
        String status,
        BigDecimal total) {
}

@JsonIgnoreProperties(ignoreUnknown = true)
public record OrderChangeEvent(
        OrderRow before,
        OrderRow after,
        String op,
        @JsonProperty("ts_ms") Long tsMs) {
}

@JsonIgnoreProperties(ignoreUnknown = true)はsourceフィールドと、Debeziumが後で追加する可能性がある新しいフィールドを無視します。Jacksonは文字列"149.90"をBigDecimalに自動変換するため、doubleを使用する場合のような精度の喪失はありません。

@Component
public class OrderChangeParser {

    private final ObjectMapper objectMapper;

    public OrderChangeParser(ObjectMapper objectMapper) {
        this.objectMapper = objectMapper;
    }

    public Optional<OrderChangeEvent> parse(String value) {
        if (value == null) {
            return Optional.empty(); // tombstone
        }
        try {
            return Optional.of(objectMapper.readValue(value, OrderChangeEvent.class));
        } catch (JsonProcessingException e) {
            throw new IllegalArgumentException("Invalid Debezium event", e);
        }
    }
}

パーサーはtombstoneにOptional.empty()を返すため、リスナーはこのケースを無視するだけです。

5.3. リスナー

@Component
public class OrderCdcListener {

    private static final Logger log = LoggerFactory.getLogger(OrderCdcListener.class);
    private final OrderChangeParser parser;

    public OrderCdcListener(OrderChangeParser parser) {
        this.parser = parser;
    }

    @KafkaListener(topics = "shop.public.orders")
    public void onOrderChange(@Payload(required = false) String value) {
        parser.parse(value).ifPresent(event -> {
            switch (event.op()) {
                case "c", "r" -> log.info("注文作成: {}", event.after());
                case "u" -> log.info("注文 {} ステータス {} -> {}",
                        event.after().id(), event.before().status(), event.after().status());
                case "d" -> log.info("注文削除: {}", event.before().id());
                default -> log.warn("不明な操作: {}", event.op());
            }
        });
    }
}

@Payload(required = false)により、Springはtombstoneを遭遇するときメソッドにnullを渡すことができます。このアノテーションがない場合、Springは変換エラーを報告します。ブランチ"r"は"c"と統合されます。消費者にとって、スナップショット内に既に存在する行は、作成されたばかりの行と同じだからです。

セクション4のデータでアプリケーションを実行すると、次のように表示されます：

INFO  OrderCdcListener : 注文作成: OrderRow[id=1, customerEmail=an@example.com, status=NEW, total=149.90]
INFO  OrderCdcListener : 注文 1 ステータス NEW -> PAID

6. パーサーのテスト

パースロジックはDebeziumやスキーマが変更される場合に最も壊れやすい部分です。Kafkaなしでこの部分をテストできます：

class OrderChangeParserTest {

    private final OrderChangeParser parser = new OrderChangeParser(new ObjectMapper());

    @Test
    void parsesUpdateEvent() {
        String json = """
            {"before":{"id":1,"customer_email":"an@example.com","status":"NEW","total":"149.90"},
             "after":{"id":1,"customer_email":"an@example.com","status":"PAID","total":"149.90"},
             "source":{"connector":"postgresql"},
             "op":"u","ts_ms":1760170000123}
            """;

        OrderChangeEvent event = parser.parse(json).orElseThrow();

        assertThat(event.op()).isEqualTo("u");
        assertThat(event.before().status()).isEqualTo("NEW");
        assertThat(event.after().status()).isEqualTo("PAID");
        assertThat(event.after().total()).isEqualByComparingTo("149.90");
    }

    @Test
    void returnsEmptyForTombstone() {
        assertThat(parser.parse(null)).isEmpty();
    }

    @Test
    void rejectsMalformedJson() {
        assertThatThrownBy(() -> parser.parse("{not json"))
                .isInstanceOf(IllegalArgumentException.class);
    }
}

最初のテストは列名customer_emailと型BigDecimalのマッピングをチェックします。2番目のテストはtombstoneがリスナーをエラーにしないことを確認します。3番目のテストは、不正なJSONが静かに飲み込まれるのではなく、明確なエラーとして報告されることを確認します。

7. よくあるエラーと使用すべきでない場合

注意： レプリケーションスロットはDebeziumが読み込みを確認するまでWALを保持します。コネクタが数日停止している場合、WALが膨張し、データベースのディスクを満杯にする可能性があります。pg_replication_slotsを監視し、コネクタを削除するときにスロットを削除する必要があります： SELECT pg_drop_replication_slot('shop_slot');

遭遇する可能性がある他の一般的なエラー：

  • コンシューマーがべき等でない。 DebeziumとKafka Connectは最低1回を保証します。再起動後、同じイベントが2回到着する可能性があります。idとsource.lsnを一緒に使って重複を避けるか、操作をupsertの形式として設計してください。
  • NUMERIC値が理解しがたいbase64文字列になった。 decimal.handling.modeの設定を忘れました。同様に、TIMESTAMPカラムはデフォルトではエポックからのマイクロ秒数として送信されるため、自分で変換する必要があります。
  • beforeが常にnull。 テーブルにREPLICA IDENTITY FULLが設定されていません。
  • 初期スナップショットが長すぎる。 大きなテーブルの場合、snapshot.mode=initialはストリーミング前にテーブル全体を読み込みます。現在からの変更のみが必要な場合はno_dataを検討するか、増分スナップショットを使用してください。
  • 内部データの漏洩。 イベントはテーブルの構造を正確に反映しています。別のサービスに直接渡すと、データベーススキーマを公開することになります。ビジネスイベントの場合、DebeziumとOutboxパターンを組み合わせてください：アプリケーションは同じトランザクション内のoutboxテーブルに書き込み、Debeziumはそのテーブルのみをキャプチャします。
  • 本番環境でpostgresアカウントを使用。 REPLICATION権限とキャプチャが必要なテーブルに対するSELECT権限を持つ専用のユーザーを作成してください。

次の場合はCDCを使用すべきではありません：

  • アプリケーションが1つだけで、同じプロセス内でイベントを自分で発行できる場合。この場合、トランザクショナルoutboxと単純なpollerで十分です。
  • “OrderPaid”のようなビジネス上の意味を持つイベントが必要な場合（“行Xのステータスカラムが変更”ではなく）。CDCは生のデータ変更を知るだけで、その理由を知りません。
  • 運用チームがKafka Connect、レプリケーションスロット、スキーマの進化を管理する準備ができていない場合。

8. まとめ

このチュートリアルでは、logical replicationを使用したPostgreSQL、Kafka、Debeziumをセットアップし、ordersテーブルをキャプチャするコネクタを登録しました。before、after、opフィールドを持つchange eventの構造を学び、tombstoneを処理するSpring Bootコンシューマーを作成し、パース部分のテストを追加しました。

CDCはデータベースが唯一の真実のソースになるため、デュアルライトのリスクを排除します。その代わりに、レプリケーションスロットを監視し、べき等なコンシューマーを作成し、明確なビジネスイベントが必要な場合はOutboxパターンの使用を検討する必要があります。

kafkadebeziumcdcpostgresqlspring-boot
著者のアプリAlterTurn photos into postage stamps, collected in a stamp bookiOS · Android

同じトピックの記事