...

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:

• Expose a new topic configuration, log.retention.min.timestamp.  The value will be a Unix time in milliseconds.

Proposed Changes

• Add a new topic configuration, log.retention.min.timestamp.
  • The format of the value will be a Unix time in milliseconds.
• Modify the log deletion mechanism (in LogManager.scala) to also delete segments whose last 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 work with both the LogAppendTime and CreateTime both CreateTime and LogAppendTime timestamp types.  There are valid use cases for timestamp-based deletion with each of these types.

...

Compatibility, Deprecation, and Migration Plan

...