Debezium-JDBC-Kafka架构下，如何删除已提交的Kafka主题消息？

首先明确：Kafka消息的删除逻辑不由Sink连接器直接控制，而是由Kafka主题自身的保留/清理策略决定，但Sink的消费偏移量提交状态会影响部分策略的生效。下面是具体解决办法：

1. 调整主题的时间/大小保留规则 #

这是最直接的方式，通过设置主题参数让Kafka自动删除过期或超出大小限制的消息：

• retention.ms: 消息的最大保留时长，比如设为86400000（24小时），超过这个时间的旧消息会被清理
• retention.bytes: 主题允许的总容量上限，达到阈值后删除最早的消息
  用Kafka命令行修改现有主题：

kafka-configs.sh --bootstrap-server <你的Kafka地址>:9092 --alter --topic <目标主题名> --add-config retention.ms=86400000,retention.bytes=1073741824

也可以在创建主题时就指定这些参数，或者修改Kafka全局配置让新主题默认套用规则。

2. 启用日志压缩（适合CDC场景） #

如果你的需求是保留每个数据库行的最新变更记录，删除旧的重复变更，可以启用日志压缩：

• 设置主题的cleanup.policy=compact，同时可以调整min.cleanable.dirty.ratio（默认0.5，控制压缩触发的脏数据比例）和delete.retention.ms（压缩后墓碑消息的保留时间）
  这种方式会让Kafka只保留每个Key的最新消息，非常适合Debezium生成的CDC主题（每个数据库行对应唯一Key）。

3. 确保Sink正确提交消费偏移量 #

JDBC Sink默认会在批量写入成功后自动提交偏移量，但你可以检查以下配置确保正常：

• auto.commit.enable: 保持默认true即可，或者用offset.flush.interval.ms调整提交频率
• 如果是手动提交模式，务必保证连接器在写入成功后正确提交偏移量，否则Kafka可能认为消息未被消费，影响清理逻辑的触发

重要提示 #

• Sink连接器没有配置项能直接强制Kafka删除消息，它只负责消费和提交偏移量，删除逻辑完全由Kafka Broker端控制
• 修改主题配置后，清理不会立即执行，需要等待Kafka后台清理线程触发（默认间隔5分钟，可通过log.cleaner.backoff.ms调整）

内容的提问来源于stack exchange，提问作者Stavros Koureas