Kafka消费者在指定的保留期之前删除消息。

这通常是由于以下两个原因之一导致的：1）Kafka broker（消息代理）的日志保留策略已经启用并且从未达到保留期限 2） 将消费者的 offsets 存储在一个不持久化的存储中，导致消费者重新启动时丢失它们的位置，从而在消费者在消费之前产生通道重置。

解决方法之一是在配置 Kafka broker 时设置保留期限。可以在 server.properties 文件中添加以下行：log.retention.hours=24 或者通过程序设置保留期限：

KafkaConfig config = new KafkaConfig(props);
config.logRetentionHours = 24;

另一种解决方法是在消费者中启用 offsets 存储。使用 Kafka 提供的 Consumer Group API 可以有效地管理 offsets。在启动消费者前，可以设置 Group ID，并将此 ID 存储在数据存储中：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");
props.put("enable.auto.commit", "false");
props.put("max.poll.records", 1);

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
    for (ConsumerRecord<String, String> record : records) {
        // process the record
        System.out.printf("offset = %d, key = %s, value = %s\n", 
                            record.offset(), record.key(), record.value());
    }
    consumer.commitSync();
}

这样，即使消费者重新启动，它也可以通过使用存储的 Group ID（在这个例子中为“test”）来重置其位置。