apache-kafka - 过期的 Apache Kafka 消息没有被删除

Question

我有一个 spring-boot (2.1.3) 服务将消息发布到 kafka(2.12-2.3.0) 主题。服务创建主题，稍后在服务启动后，将 retention.ms 设置为 1 秒。

正在调试这段代码

@SpringBootApplication()
@EnableAsync
public class MetricsMsApplication {

  public static void main(String[] args) {
    SpringApplication.run(MetricsMsApplication.class, args);
}

@Bean
public NewTopic topic1() {

    NewTopic nt = new NewTopic("metrics", 10, (short) 1);
    return nt;
}

@EventListener(ApplicationReadyEvent.class)
private void init() throws ExecutionException, InterruptedException {
    Map<String, Object> config = new HashMap<>();
    config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");

    AdminClient client = AdminClient.create(config);

    ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, "metrics");

    // Update the retention.ms value
    ConfigEntry retentionEntry = new ConfigEntry(TopicConfig.RETENTION_MS_CONFIG, "1000");
    Map<ConfigResource, Config> updateConfig = new HashMap<ConfigResource, Config>();
    updateConfig.put(resource, new Config(Collections.singleton(retentionEntry)));
    AlterConfigsResult alterConfigsResult = client.alterConfigs(updateConfig);
    alterConfigsResult.all();
}

我发送几条消息并数到 5，然后启动控制台消费者

kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic admst-metrics --from-beginning

并且仍然得到应该已经过期的消息。

kafka 日志显示已应用 retention.ms 配置。我添加了 cleanup.policy 并将其设置为删除，但这不是必需的，因为它是默认设置。

什么会导致这些消息被删除？

Answer 1

简短的回答 - kafka 并不是为了满足如此低的保留值而设计的。

较长的答案:

Kafka 将任何(主题)分区的数据存储在段文件中。在任何时候，一个段都是“事件的”并被写入，而所有较旧的段都是“关闭的”。保留/压缩仅适用于非事件段。

当 log.roll.ms 或 log.segment.bytes 被命中时，Kafka 会滚动新的段。默认值(参见 https://kafka.apache.org/documentation/#brokerconfigs )为 7 天和/或 ~1GB。

还有 log.segment.delete.delay.ms，默认情况下这意味着任何段至少保留一分钟。

压缩/删除非事件段的工作由日志清理器线程完成。当没有找到任何工作时，那些 sleep log.cleaner.backoff.ms(15 秒)并且只检查每个 log.retention.check.interval.ms(5 分钟)是否可以清理任何特定段

所有这一切的结果是，默认情况下，保留值不可能接近您要查找的值。

您可以尝试调整上述所有值，看看可以降低到多低，但我敢打赌这不会很好地扩展到大量主题。