...

A new addition to the existing deletion policy based on the absolute timestamp of a message would work perfectly for this case.  The client application will periodically update the minimum timestamp of messages to retain, and Kafka will delete all messages earlier than that timestamp using the existing log deletion mechanism, alongside the existing size-based and duration-based checks.

This is based off of the work being done in for KIP-32 - Add timestamps to Kafka message. and KIP-33 - Add a time based log index.

Public Interfaces

This KIP has the following public interface changes:

...

• Add a new topic configuration, log.retention.min.timestamp
• Modify the log deletion mechanism (in LogManager.scala) to also delete segments whose last message timestamp is before the configured timestamp if the timestamp is set
  • This check will use the time-based log index from KIP-33 - Add a time based log index.
• Timestamp-based deletion will only work with the both CreateTime and LogAppendTime timestamp type.
  • LogAppendTime guarantees that timestamps are sequentially increasing, and so it is impossible to have a segment which has a last message with an earlier timestamp than another message in the segment.
  • CreateTime could be used if we check every single message in a segment and take the max timestamp, but this would be significantly slower than just checking the last message.

...

• types.

Compatibility, Deprecation, and Migration Plan

...