Kafka数据保留策略基于时间（log.retention.ms）和大小（log.retention.bytes），可单独或组合使用，配合log.cleanup.policy设置delete或compact策略，实现过期数据清理。
Kafka的数据保留策略主要基于时间和大小两种维度。时间维度是指数据在Kafka集群中保留的最长时间，超过这个时间的数据会被删除。大小维度是指每个Topic或者Partition允许存储的最大数据量，当数据量超过这个限制时，Kafka会根据配置的策略删除旧的数据。这两种策略可以单独使用，也可以组合使用，以满足不同的业务需求。
解决方案 Kafka的数据保留策略主要通过以下几个配置项来控制：
log.retention.ms
: 控制消息保留的最长时间，单位是毫秒。例如，设置为604800000
表示保留7天。log.retention.bytes
: 控制每个Partition允许保留的最大数据量，单位是字节。例如，设置为1073741824
表示保留1GB。log.cleanup.policy
: 控制数据清理策略，可选值为delete
和compact
。delete
表示直接删除过期数据，compact
表示对数据进行压缩，只保留每个Key的最新值。
具体来说，Kafka会定期检查每个Partition的数据，判断是否满足删除条件。检查的频率由
log.retention.check.interval.ms配置项控制，默认值为300000毫秒（5分钟）。
当
log.clean设置为up.policy
delete时，Kafka会直接删除满足以下任一条件的数据：
- 消息的写入时间超过
log.retention.ms
设置的值。 - Partition的总数据量超过
log.retention.bytes
设置的值。
当
log.cleanup.policy设置为
compact时，Kafka会定期对Partition的数据进行压缩，只保留每个Key的最新值。压缩的过程由Log Compactor线程负责，它会扫描Partition的数据，找出重复的Key，然后只保留最新的值。压缩的频率由
log.cleaner.enable和
log.cleaner.threads配置项控制。需要注意的是，
compact策略需要依赖Key的存在，并且会增加Kafka的CPU和IO负载。
如果同时设置了
log.retention.ms和
log.retention.bytes，Kafka会优先满足先达到的条件。例如，如果设置
log.retention.ms=604800000（7天）和
log.retention.bytes=1073741824（1GB），那么当Partition的数据量达到1GB或者消息的写入时间超过7天时，Kafka就会开始删除数据。
实际应用中，需要根据业务需求选择合适的数据保留策略。例如，对于需要长期保存的数据，可以选择较大的
log.retention.ms和
log.retention.bytes值；对于只需要保留最新状态的数据，可以选择
compact策略。
Kafka如何处理消息过期但未被立即删除的情况？
即使消息满足了删除条件，Kafka也不会立即删除它们。这是因为Kafka的数据存储结构是基于日志的，删除操作会影响性能。Kafka采用了一种批量删除的策略，它会定期扫描日志文件，找出可以删除的消息，然后一次性删除。这个过程由Log Cleaner线程负责。
具体来说，Kafka会将日志文件分成多个Segment，每个Segment包含一部分消息。当Segment中的所有消息都满足删除条件时，Kafka才会删除整个Segment文件。如果Segment中只有部分消息满足删除条件，Kafka会将这些消息标记为已删除，但不会立即删除它们。这些被标记为已删除的消息会在后续的压缩过程中被清理掉。
因此，即使消息已经过期，它们仍然会占用一定的存储空间，直到被Log Cleaner线程清理掉。这可能会导致实际的存储空间使用量超过
log.retention.bytes设置的值。
此外，如果消费者消费速度慢于生产者生产速度，也可能导致消息堆积，从而加速消息的过期。为了避免这种情况，可以考虑增加消费者数量，或者优化消费者的消费逻辑。
如何动态修改Kafka的保留策略？
Kafka允许动态修改Topic级别的保留策略，而无需重启Broker。这可以通过Kafka的
kafka-configs.sh脚本或者Kafka AdminClient API来实现。
例如，要将Topic "my-topic"的保留时间修改为14天，可以使用以下命令：
./kafka-configs.sh --zookeeper localhost:2181 --alter --entity-type topics --entity-name my-topic --add-config retention.ms=1209600000
这个命令会将
retention.ms配置项的值修改为1209600000毫秒（14天）。
需要注意的是，动态修改配置项可能会影响Kafka的性能。例如，如果将保留时间设置得过短，Kafka会频繁地删除数据，从而增加IO负载。因此，在修改配置项之前，应该仔细评估其对Kafka性能的影响。
另外，动态修改配置项只会影响新写入的数据，不会影响已经存在的数据。如果需要修改已经存在的数据的保留策略，需要手动删除这些数据。
除了时间和大小，还有没有其他的数据保留策略？
除了时间和大小，Kafka还可以根据消息的Offset来保留数据。这可以通过设置
log.segment.bytes和
log.segment.ms配置项来实现。
log.segment.bytes控制每个Segment文件的大小，当Segment文件达到这个大小时，Kafka会创建一个新的Segment文件。
log.segment.ms控制Segment文件的创建时间，当Segment文件创建时间超过这个时间时，Kafka也会创建一个新的Segment文件。
通过控制Segment文件的大小和创建时间，可以间接地控制消息的Offset范围。例如，如果设置
log.segment.bytes=1073741824（1GB）和
log.segment.ms=604800000（7天），那么每个Segment文件最多包含1GB的数据，或者最多保留7天的数据。
当需要根据Offset来保留数据时，可以先计算出需要保留的Offset范围，然后根据这个范围来设置
log.segment.bytes和
log.segment.ms的值。
需要注意的是，根据Offset来保留数据可能会导致数据丢失。例如，如果某个Segment文件中的消息的Offset范围不连续，那么删除这个Segment文件可能会导致部分消息丢失。因此，在使用这种策略时，需要仔细评估其对数据完整性的影响。
