使用 SCS 删除消费消息的 kafka 日志

Question

我是使用 kafka 和 Spring Cloud Stream 的新手。需要一些帮助。

设置

• 我有两个 spring-boot 应用程序 App-1、App-2。
• 我正在使用 spring cloud stream 和 spring-cloud-stream-binder-kafka 进行异步通信。
• 有一个主题TOPIC-1

---

用例

• 假设 App-1 发送了一条关于 App-2 正在侦听的主题 TOPIC-1 的消息。
• App-2 使用该消息并成功处理它。
• 现在该主题的偏移量增加了。

  ---

问题

• 如何实现一种机制，在指定时间段后从 kafka 日志中删除唯一成功消费的消息数据？

在 Kafka 中，对已消费的内容的责任是消费者的责任。所以我想，spring cloud stream kafka中一定有一些我不知道的kafka消息日志控制机制。

---

注意 1：我知道 kafka 日志保留时间和磁盘属性。但即使是未消费的消息，kafka 日志也会被删除。

注意 2：我已经通过 this question 但它无济于事。

Answer 1

据我所知，在 Kafka 中没有这样的机制；当然不在 Spring Cloud Stream 或它所基于的库中。 Kafka 客户端无法访问此类低级构造。

此外，消费者偏移量与主题日志完全分开；在现代经纪人中，它们被存储在一个特殊的主题中。

编辑

根据下面的评论，可以使用kafka-delete-records.sh 命令行工具。

请注意，这使用默认情况下不在 SCSt 类路径上的 scala AdminClient（自 2.0 起）。

不过，java AdminClient 支持类似的功能：

/**
 * Delete records whose offset is smaller than the given offset of the corresponding partition.
 *
 * This is a convenience method for {@link #deleteRecords(Map, DeleteRecordsOptions)} with default options.
 * See the overload for more details.
 *
 * This operation is supported by brokers with version 0.11.0.0 or higher.
 *
 * @param recordsToDelete       The topic partitions and related offsets from which records deletion starts.
 * @return                      The DeleteRecordsResult.
 */
public DeleteRecordsResult deleteRecords(Map<TopicPartition, RecordsToDelete> recordsToDelete) {
    return deleteRecords(recordsToDelete, new DeleteRecordsOptions());
}

您可以使用引导的AutoConfiguration KafkaAdmin 创建一个AdminClient。

AdminClient client = AdminClient.create(kafkaAdmin.getConfig());