...

• 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 CreateTime and LogAppendTime timestamp types.

...