Kafka日志没有按照给定保留时间存储。

1. 确保Kafka的服务端配置中设置了正确的日志保留时间，例如设置为7天：

log.retention.hours=168

2. 检查Kafka的日志清理策略，确保日志清理器没有被禁用，并且Kafka配置中设置了正确的日志清理策略。例如，以下配置将启用按照时间保留和基于大小的保留：

log.cleanup.policy=delete log.retention.ms=604800000 log.segment.bytes=1073741824

3. 检查Kafka的磁盘空间，确保磁盘空间足够存储应该保留的所有日志。

4. 如果您在使用Kafka Streams，检查您的应用程序中是否存在任何异常，例如未处理的异常，这可能会影响Kafka的日志保留和清理。

代码示例：

以下是在Kafka中设置日志保留时间和清理策略的示例代码：

Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", "test-group"); props.put("enable.auto.commit", "true"); props.put("auto.commit.interval.ms", "1000"); props.put("session.timeout.ms", "30000"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

// 设置日志保留时间 props.put("log.retention.hours", "168");

// 设置日志清理策略 props.put("log.cleanup.policy", "delete"); props.put("log.retention.ms", "604800000"); props.put("log.segment.bytes", "1073741824");

// 创建Kafka消费者 KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

// 订阅一个或多个主题 consumer.subscribe(Arrays.asList("test-topic"));

// 消费数据 while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); } }

注：在您的Kafka安装中，可能需要调整一些属性的值来正确地设置保留时间和清理策略。