...

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 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.

Note: This Change focuses on when to compact a log segment, and it doesn’t conflict with KIP-280, which focuses on how to compact log.

Current Behavior

For log compaction enabled topic, Kafka today uses “min.cleanable.dirty.ratio” and "min.compaction.lag.ms" to determine what log segments it need to pick up for compaction. "min.compaction.lag.ms" marks a log segment uncleanable until the segment is sealed and remains un-compacted for the specified "lag". The detailed information can be found in KIP-58.   “min.cleanable.dirty.ratio” is used to determine the eligibility of the entire partition for log compaction. Only log partitions whose dirty ratio is higher than “min.cleanable.dirty.ratio” are picked up by log cleaner for compaction.  In addition, when log cleaner performs compaction on a log partition, there is no guarantee it will compact all cleanable segments determined by "min.compaction.lag.ms". On each compaction run, log cleaner will build an offsetmap, the number of records that can be fit into the offsetmap also limit the number of log segments that can be compacted. In summary, with these two compaction configurations,  Kafka cannot enforce a timely log compaction.

Proposed Changes

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 lag. The clock starts when a log segment is first created as an active segment.

...

• 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. This Change focuses on when to compact a log segment, and it doesn’t conflict with KIP-280, which focuses on how to compact log. 

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.

...