Kafka保留策略不按预期工作”

1. 确保在broker级别正确配置了retention.ms或retention.bytes参数。例如：

broker: log.retention.ms=86400000 # 1 day log.retention.bytes=1073741824 # 1GB

2. 如果在topic级别上配置了保留时间或者大小，则确保它们的值小于或等于broker级别的配置值。例如：

kafka-topics --zookeeper localhost:2181 --alter --topic my_topic --config retention.ms=86400000 kafka-topics --zookeeper localhost:2181 --alter --topic my_topic --config retention.bytes=1073741824

3. 确保在使用KafkaProducer发送数据时正确设置了时间戳。例如：

producer.send(new ProducerRecord<String, String>("my_topic", System.currentTimeMillis(), "my_key", "my_value"));

4. 如果使用KafkaStreams进行流处理，则确保在应用程序中正确设置了保留策略。例如：

final StreamsBuilder builder = new StreamsBuilder(); final KStream<String, String> stream = builder.stream("my_topic"); stream.groupByKey().windowedBy(TimeWindows.of(Duration.ofSeconds(30))) .reduce((a, b) -> a + b) .toStream().to("my_output_topic", Produced.with(Serdes.String(), Serdes.String()) .withRetention(Duration.ofMinutes(5)));

以上是一些可能会导致Kafka保留策略不按预期工作的原因及对应的解决方案。