分布式环境下Spring整合Debezium Embedded如何用Kafka存储offset？

当然可以用Kafka来跟踪Debezium嵌入式连接器的offset，完全不需要依赖数据库或本地文件，非常适配你的分布式环境。你只需要替换当前配置中offsetStorageFileName相关参数，改用Kafka-backed的offset存储实现即可。

核心配置修改步骤 #

1. 替换Offset存储类型：指定Kafka作为offset的持久化实现类
2. 配置Kafka集群参数：设置Kafka地址、offset存储Topic、消费组ID
3. 可选优化：同步数据库历史到Kafka：分布式环境下建议把数据库schema变更历史也存到Kafka，避免本地文件不一致问题

修改后的配置代码示例 #

调整buildDebeziumConnectorString方法，替换原有的offset和历史文件配置：

public String buildDebeziumConnectorString() {
    StringBuilder sb = new StringBuilder(SQLSERVER_CONNECTOR_NAME);
    sb.append("?databaseHostName=").append(properties.getDatabaseHostName())
      .append("&databasePort=").append(properties.getDatabasePort())
      .append("&databaseUser=").append(properties.getDatabaseUser())
      .append("&databasePassword=").append(properties.getDatabasePassword())
      .append("&databaseServerName=").append(properties.getDatabaseServerName())
      .append("&includeSchemaChanges=").append(properties.isIncludeSchemaChanges())
      .append("&databaseDbname=").append(properties.getDatabaseDbname())
      .append("&tableWhitelist=").append(properties.getTableWhitelist())
      // 替换offset存储为Kafka
      .append("&offsetStorage=org.apache.kafka.connect.storage.KafkaOffsetBackingStore")
      .append("&offsetStorageKafkaBootstrapServers=").append(properties.getKafkaBootstrapServers())
      .append("&offsetStorageKafkaTopic=").append(properties.getOffsetStorageTopic()) // 示例："debezium-sqlserver-offsets"
      .append("&offsetStorageKafkaGroupId=").append(properties.getOffsetGroupId()) // 示例："debezium-sqlserver-group"
      // 可选：将数据库历史也存储到Kafka（分布式环境必备）
      .append("&databaseHistory=io.debezium.relational.history.KafkaDatabaseHistory")
      .append("&databaseHistoryKafkaBootstrapServers=").append(properties.getKafkaBootstrapServers())
      .append("&databaseHistoryKafkaTopic=").append(properties.getDatabaseHistoryTopic()); // 示例："debezium-sqlserver-history"
    return sb.toString();
}

关键参数说明 #

• offsetStorage：固定值org.apache.kafka.connect.storage.KafkaOffsetBackingStore，指定使用Kafka存储offset
• offsetStorageKafkaBootstrapServers：Kafka集群地址，格式为host1:port1,host2:port2
• offsetStorageKafkaTopic：存储offset的专属Topic，默认是connect-offsets，建议自定义避免冲突
• offsetStorageKafkaGroupId：连接器消费组ID，分布式环境下同一数据源的连接器必须用相同GroupId，才能共享offset状态
• databaseHistory：将schema变更历史存到Kafka，解决多实例间历史文件不一致的问题

注意事项 #

• 确保Kafka集群正常运行，且连接器拥有对应Topic的读写权限
• 生产环境建议提前手动创建offset和历史Topic，并配置合理的分区数与副本数；测试环境可开启Kafka的auto.create.topics.enable自动创建
• 分布式部署时，所有连接同一SQL Server的Debezium节点，必须使用相同的databaseServerName、offsetStorageKafkaTopic和offsetStorageKafkaGroupId，否则会出现重复消费或offset不一致问题

内容的提问来源于stack exchange，提问作者AndreaCavallo.class