从ActiveMQ迁移至Kafka：仅删已消费消息、查看未消费消息及问题

我们正从ActiveMQ迁移到Kafka。在ActiveMQ中，我们向多个队列写入大量消息，消息被消费后会自动删除，仅未消费消息保留在队列中。我们希望在Kafka中实现相同行为——不保留数百万条已消费消息，只保留未消费消息。

我配置了Kafka的log.retention.hours=1属性，试图尽快删除所有已消费消息。

测试步骤 #

1. 创建主题crs.reporting.intern，发送2条消息并成功消费，状态正常：

C:\Java\apache-kafka_2.13-3.3.1\bin\windows>kafka-consumer-groups.bat --bootstrap-server localhost:9092 --describe --group groupid.crs.reporting.intern
GROUP                        TOPIC                PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                                                   HOST            CLIENT-ID
groupid.crs.reporting.intern crs.reporting.intern 0          2               2               0               consumer-groupid.crs.reporting.intern-12-8fc98e12-d884-4c59-b41c-768721f98e51 /172.31.0.1     consumer-groupid.crs.reporting.intern-12

2. 停止消费者，此时无活跃消费者监听该主题：

C:\Java\apache-kafka_2.13-3.3.1\bin\windows>kafka-consumer-groups.bat --bootstrap-server localhost:9092 --describe --group groupid.crs.reporting.intern
Consumer group 'groupid.crs.reporting.intern' has no active members.
GROUP                        TOPIC                PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
groupid.crs.reporting.intern crs.reporting.intern 0          2               2               0               -               -               -

3. 在无活跃消费者的情况下发送第三条消息，但查看消费组状态与步骤2一致：

C:\Java\apache-kafka_2.13-3.3.1\bin\windows>kafka-consumer-groups.bat --bootstrap-server localhost:9092 --describe --group groupid.crs.reporting.intern
Consumer group 'groupid.crs.reporting.intern' has no active members.
GROUP                        TOPIC                PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
groupid.crs.reporting.intern crs.reporting.intern 0          2               2               0               -               -               -

4. 等待超过1小时后启动消费者，发现3条消息均被删除（包括2条已消费消息和1条未消费消息）：

C:\Java\apache-kafka_2.13-3.3.1\bin\windows>kafka-consumer-groups.bat --bootstrap-server localhost:9092 --describe --group groupid.crs.reporting.intern
GROUP                        TOPIC                PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                                                   HOST            CLIENT-ID
groupid.crs.reporting.intern crs.reporting.intern 0          2               2               0               consumer-groupid.crs.reporting.intern-60-60a4ffb9-7dea-44c6-a34e-7fb52ef61223 /172.31.0.1     consumer-groupid.crs.reporting.intern-60

疑问 #

• 问题3：如何在第三条未消费消息被删除前查看它？
• 问题4：有没有办法防止未消费消息被删除？在此场景下，我预期仅前两条已消费消息被删除，而非第三条！

消息流向：[Data]-send->[Kafka Producer]-send->[**Kafka Broker**]<-Read-[Kafka Consumer]-Write->[Data Store]，按道理第三条消息应该仍在Kafka Broker中，但实际已被删除。

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