Kafka Log Compaction实践问题：消息丢失与删除压缩实现求助

先来说你遇到的第一个问题：为什么不同Key的消息会丢失

从你的操作步骤和配置来看，主要有两个核心原因：

1. 生产者异步发送未等待确认就关闭
   你的Scala生产者代码里，producer.send(record)是异步操作，没有等待发送请求完成就直接调用了producer.close()。这会导致一部分消息还没来得及发送到Kafka集群，生产者就被终止，自然不会被消费者接收到。
   修复方法是等待所有发送请求完成后再关闭生产者：

   val data_amount = 100 
   val topic = "COMPACTION10" 
   val kafkaConfiguration = new Properties 
   kafkaConfiguration.put("bootstrap.servers", "localhost:9092") 
   kafkaConfiguration.put("key.serializer", Class.forName("org.apache.kafka.common.serialization.StringSerializer")) 
   kafkaConfiguration.put("value.serializer", Class.forName("org.apache.kafka.common.serialization.StringSerializer")) 
   val producer = new KafkaProducer[String, String](kafkaConfiguration) 
   
   // 收集所有发送请求的Future
   val sendFutures = (1 to data_amount).map { id =>
     println(id)
     val record = new ProducerRecord(topic, id.toString, id.toString)
     producer.send(record)
   }
   
   // 等待所有请求完成，确保消息都发送成功
   sendFutures.foreach(_.get())
   producer.close()
   

2. 过于激进的日志段配置引发异常
   你设置的segment.bytes=1000（每个日志段仅1KB）和segment.ms=100（每100毫秒滚动一个新段）会生成大量极小的日志段文件。Kafka的后台清理线程和段管理线程无法及时处理这么频繁的段滚动，甚至可能出现段文件未完全写入就被切换的情况，导致部分消息丢失。
   建议先把这些参数调整为合理值，比如segment.bytes=10485760（10MB）、segment.ms=3600000（1小时），再测试是否还会出现消息丢失。

另外还要确认消费者配置：如果你的消费者没有设置auto.offset.reset=earliest，首次启动或offset不存在时会默认从最新位置消费，这也会导致你看不到之前发送的历史消息。

再来说第二个问题：如何发送空值实现删除压缩（墓碑消息）

Kafka的Log Compaction中，要触发Key的删除，需要发送墓碑消息（Tombstone Message）——也就是value为null的消息，而不是空字符串""。空字符串是有效的字符串值，Kafka会把它当作普通消息处理，不会触发删除逻辑。

在Scala中，你可以直接创建value为null的ProducerRecord，StringSerializer原生支持序列化null值：

// 发送墓碑消息删除指定Key
val tombstoneRecord = new ProducerRecord[String, String](topic, "1", null)
// 等待发送完成确保墓碑消息写入
producer.send(tombstoneRecord).get()

发送墓碑消息后，Log Cleaner执行压缩时会保留这条墓碑消息；等到delete.retention.ms（你设置的100/10000毫秒）超时后，才会彻底删除该Key的所有记录（包括墓碑消息）。所以你可能会先看到value为null的消息，过一段时间后才会完全看不到这个Key的痕迹。

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