配置Debezium CDC与所有扩容Postgres副本实例的Sink连接

背景 #

• 主数据库：作为Debezium捕获变更的源数据库
• 副本数据库：通过Docker-Compose扩容的多个主库克隆实例
• Debezium-Kafka CDC：用于捕获并发布数据库变更

问题详情 #

主库发生变更时，Debezium的Sink连接器仅连接其中一个副本库，导致同步流程出现瓶颈，期望实现Sink连接器并行分发变更至所有副本库。

已尝试操作 #

尝试多种配置均无法实现Debezium同时连接所有副本库，推测Docker-Compose的负载均衡是问题根源，但不知如何解决。

附简化配置 #

Docker-Compose配置 #

version: '3'
services:
  primary-db:
    image: my-primary-db-image
    ports:
      - "5432:5432"
    # ...

  replica-db:
    image: my-replica-db-image
    # ...

  debezium:
    image: debezium/connect
    # ...

Debezium源连接器配置 #

{
  "name": "my-source-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "primary-db",
    "database.port": "5432",
    "database.user": "user",
    "database.password": "password",
    "database.dbname": "mydb",
    # ... 其他源连接器配置
  }
}

1. 禁用Docker-Compose默认负载均衡，为每个副本分配唯一网络标识 #

Docker-Compose中同一服务名的多个实例会被默认负载均衡，导致Debezium只能解析到一个实例地址。需要为每个副本显式配置不同的服务名：

修改后的Docker-Compose配置：

version: '3'
services:
  primary-db:
    image: my-primary-db-image
    ports:
      - "5432:5432"
    # ...

  replica-db-1:
    image: my-replica-db-image
    networks:
      - db-network
    # ...

  replica-db-2:
    image: my-replica-db-image
    networks:
      - db-network
    # ...

  debezium:
    image: debezium/connect
    networks:
      - db-network
    # ...

networks:
  db-network:
    driver: bridge

若需动态扩容，可配合docker compose up --scale replica-db=N，但需额外引入服务发现组件（如Consul）让Debezium识别所有副本地址；静态配置独立服务名是最直接稳定的方式。

2. 为每个副本创建独立的Debezium Sink连接器 #

每个副本对应一个独立的Sink连接器，确保变更被并行推送到所有实例：

副本1的Sink连接器配置：

{
  "name": "my-sink-replica-1",
  "config": {
    "connector.class": "io.debezium.connector.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "topics": "dbserver1.mydb.public.*",
    "connection.url": "jdbc:postgresql://replica-db-1:5432/mydb",
    "connection.user": "user",
    "connection.password": "password",
    "insert.mode": "upsert",
    "delete.enabled": "true",
    "table.name.format": "${topic.schema}.${topic.table}",
    # ... 其他Sink配置
  }
}

副本2的Sink连接器配置：

{
  "name": "my-sink-replica-2",
  "config": {
    "connector.class": "io.debezium.connector.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "topics": "dbserver1.mydb.public.*",
    "connection.url": "jdbc:postgresql://replica-db-2:5432/mydb",
    "connection.user": "user",
    "connection.password": "password",
    "insert.mode": "upsert",
    "delete.enabled": "true",
    "table.name.format": "${topic.schema}.${topic.table}",
    # ... 其他Sink配置
  }
}

3. 优化Kafka主题分区提升并行性 #

若主库变更量较大，可将源连接器的主题分区数与副本数量匹配，让每个Sink连接器消费指定分区：

修改源连接器配置，增加分区设置：

{
  "name": "my-source-connector",
  "config": {
    // ... 原有配置
    "topic.creation.default.partitions": "2", // 与副本数量一致
    "topic.creation.default.replication.factor": "1"
  }
}

在Sink连接器中指定消费策略（可选）：

{
  "name": "my-sink-replica-1",
  "config": {
    // ... 原有配置
    "topic.partition.assignment.strategy": "org.apache.kafka.clients.consumer.RoundRobinAssignor",
    "consumer.group.id": "sink-group-replica-1"
  }
}

4. 动态副本的替代方案（可选） #

若需支持动态扩容且避免手动创建Sink连接器，可：

• 用Kafka Streams实现变更广播，结合服务发现组件拉取实时副本列表
• 自定义Sink连接器逻辑，自动遍历所有副本地址并行推送变更

此方案需额外开发或依赖第三方组件，静态配置多个Sink连接器仍是最易维护的方案。

内容的提问来源于stack exchange，提问作者Hao Phan Phu