Kafka Topic压缩（compact）功能失效问题排查求助

问题核心 #

你配置了Kafka日志压缩但仍出现相同Key的重复记录，核心问题出在配置的错误放置、已存在Topic的策略未更新，以及压缩触发条件的理解偏差上。

配置问题排查与修复 #

1. 修正Broker级配置的位置 #

你把log.cleaner.*这类Kafka Broker全局配置错误地放在了Debezium Connector配置中，这些配置必须放在Kafka Broker的server.properties文件里，而非Connector配置。正确的Broker配置如下：

log.cleaner.enable=true
log.cleaner.delete.retention.ms=1000
log.cleaner.min.cleanable.ratio=0.1
log.cleaner.min.compaction.lag.ms=1000

修改后重启Kafka Broker。

2. 更新已存在Topic的清理策略 #

如果目标Topic在配置topic.creation.default.cleanup.policy=compact之前已经创建，默认配置不会自动覆盖已存在的Topic。需要手动修改Topic的清理策略：

kafka-configs.sh --bootstrap-server kafka:9092 --alter --topic paraph.database.<你的表名> --add-config cleanup.policy=compact

替换<你的表名>为实际数据库表名，若需批量更新Debezium生成的所有Topic，可使用通配符（需Kafka支持）。

3. 确认Debezium消息Key的正确性 #

日志压缩依赖相同Key的记录去重，需确保：

• 数据库表存在主键（无主键的表无法触发有效压缩）
• 当前Connector的transforms.unwrap配置正确：io.debezium.transforms.ExtractNewRecordState不会破坏主键作为Kafka消息Key的逻辑，无需调整。

4. 手动触发压缩验证 #

若需立即验证效果，可强制日志段滚动触发清理：

kafka-run-class.sh kafka.admin.LogDirsCommand --bootstrap-server kafka:9092 --alter --topic paraph.database.<你的表名> --set log.segment.bytes=1

该命令会让当前日志段立即滚动，Kafka清理线程随即开始压缩处理。

验证压缩效果 #

使用以下命令查看Topic内容，确认每个Key仅保留最新记录：

kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic paraph.database.<你的表名> --from-beginning --property print.key=true --property key.separator=":"

内容的提问来源于stack exchange，提问作者ahlam el