...

Compaction enables Kafka to remove old messages that are flagged for deletion while other messages can be retained for a relatively longer time.  Today, a log segment may remain un-compacted for a long time since the eligibility for log compaction is determined based on compaction ratio (“min.cleanable.dirty.ratio”) and min compaction lag ("min.compaction.lag.ms") setting.  Ability to delete a log message through compaction in a timely manner has become an important requirement in some use cases (e.g., GDPR).  The   For example,  one use case is to delete PII (Personal Identifiable information) data within 7 days while keeping non-PII indefinitely in compacted format.  The goal of this change is to provide a time-based compaction policy that ensures the cleanable section is compacted after the specified time interval regardless of dirty ratio and “min compaction lag”.  However, dirty ratio and “min compaction lag” are still honored if the time based compaction rule is not violated. In other words, if Kafka receives a deletion request on a key (e..g, a key with null value), the corresponding log segment will be picked up for compaction after the configured time interval to remove the key.

...

We propose adding a new topic level configuration: “max.compaction.lag.ms”, which controls the max time interval a message/segment can be skipped for log compaction (note that this interval includes the time the message resides in an active segment).  With this configuration and compaction enabled, log cleaner is required to pick up all log segments that contain messages older than “max.compaction.lag.ms” for compaction. A log segment has a guaranteed upper-bound in time to become mandatory for compaction despite compaction ratio and min compaction lagmin cleanable dirty ratio. The clock starts when a log segment is first created as an active segment.

...

• Adding topic level configuration "max.compaction.lag.ms",  and corresponding broker configuration "log.cleaner.max.compaction.lag.ms", which is set to 0 (disabled) by default. "segmentKafka enforces "max.compaction.lag.ms" : no change in meaning.  However, if "maxis no less than "min.compaction.lag.ms".

• "segment.ms" : no change in meaning.  is set to a non-zero smaller value than segment.ms, segment.ms doesn't really control the rolling time. The active segment is forced to roll when either "max.compaction.lag.ms" or "segment.ms" (log.roll.ms and log.roll.hours) has reached.  We recommend setting "max.compaction.lag.ms" to a value greater than "segment.ms" in practice.  

• min.cleanable.dirty.ratio : no change in meaning. However, the compaction decision that made based on "max.compaction.lag.ms" will override the compaction decision made based on "min.cleanable.dirty.ratio".

• min.compaction.lag.ms : no change in meaning. We recommend setting "min.compaction.lag.ms"  has to be a smaller value than or equal to "max.compaction.lag.ms".  If not, "max.compaction.lag.ms" has higher priority.  

• All above changes are only applicable for topics when compaction is enabled.

...

• By default "max.compaction.lag.ms" is set to 0 and this time-based log compaction policy is disabled.  There are no compatibility issues and no migration is required.   

Rejected Alternatives

• One way to force compaction on any cleanable log segment is setting “min.cleanable.dirty.ratio” to 0. However, compacting a log partition whenever a segment become cleanable (controlled by "min.compaction.lag.ms") is very expensive.  We still want to accumulate some amount of log segments before compaction is kicked out.Prefer "min.compaction.lag.ms" over "max.compaction.lag.ms".  We decide not to honor "min.compaction.lag.ms" if "min.compaction.lag.ms" is conflict with "max.compaction.lag.ms". The reason is that we need to provide a stronger guarantee that deletion request can be fulfilled on time.  For example, if "min.compaction.lag.ms" is set to two days and "max.compaction.lag.ms" is set to one day, the log will not be compacted in the first day but the log will be compacted after first day.  However, this conflict setting may lead to frequent compaction since there is no time interval to accumulate cleanable segments. Considering the previous example, a log segment is uncleanable in the first day, but it will trigger a compaction as soon as it enters the second day time window.  Due to this reason, we strongly recommend to set "max.compaction.lag.ms" higher than "min.compaction.lag.ms", and leave some time interval in between to accumulate some cleanable log segments that can be compacted in a single compaction run.

• If compaction and time based retention are both enabled on a topic, the compaction might prevent records from being deleted on time.  The reason is when compacting multiple segments into one single segment, the newly created segment will have same lastmodified timestamp as latest original segment. We lose the timestamp of all original segments except the last one. As a result, records might not be deleted as it should be through time based retention.  We decide not to address this issue in this KIP because  we don't have obvious use cases that users must enable both time based retention and log compaction. Addressing this issue can be kept as a future work.  One solution is during log compaction, looking into record timestamp to delete expired records. This can be done in compaction logic itself or use AdminClient.deleteRecords() . But this solution assumes we have record timestamp.  Further investigation is needed if we have to deal with on-time retention on log compacted topic.