...

We propose adding a new topic level configuration: “max.compaction.lag.ms”, which controls the max lag after which a record is required to be picked up for compaction (note that this lag interval includes the time the record resides in an active segment). In other words, a message record has a guaranteed upper-bound in time to become mandatory for compaction despite min cleanable dirty ratio. The clock starts  The lag is measured starting from when the message record is appended to an active segment. 

Here are a list of changes to enforce such a max compaction lag:

1. 1. Force a roll of non-empty active segment if the first record is older than "max.compaction.lag.ms"  so that compaction can be done on that segment.  The time to roll an active segments is controlled by "segment.ms" today.  However, to ensure messages currently in the active segment can be compacted in time, we need to roll the active segment when either "max.compaction.lag.ms" or "segment.ms" is reached.  
   We define: 
   

   Info
   icon false title maximum time to roll an active segment:
   maxSegmentMs =  if (the log has "compact" enabled) {min(“segment.ms”, “max.compaction.lag.ms")}                                 else {“segment.ms” }

   
   
   

2. Estimate the earliest message timestamp of an un-compacted log segment. we only need to estimate earliest message timestamp for un-compacted log segments to ensure timely compaction because the deletion requests that belong to compacted segments have already been processed.

   1. for the first (earliest) log segment:  The estimated earliest timestamp is set to the timestamp of the first message if timestamp is present in the message. Otherwise, the estimated earliest timestamp is set to "segment.largestTimestamp - maxSegmentMs”  (segment.largestTimestamp is lastModified time of the log segment or max timestamp we see for the log segment.). In the later case, the actual timestamp of the first message might be later than the estimation, but it is safe to pick up the log for compaction earlier.  

   2. from the second log segment onwards:  there are two methods to estimate the earliest message timestamp of a log segment. First method is to   we can use the largestTimestamp (lastmodified time) of previous segment as an estimation. Second method is to use the timestamp of the first message if timestamp is present in the message.  Since getting the timestamp of a message requires additional IOs, the first method of estimation is sufficient in practice.

3. Let log cleaner pick up logs that have reached max compaction lag for compaction.  
   For any given log partition with compaction enabled, as long as the estimated earliest message timestamp of first un-compacted segment is earlier than "max.compaction.lag.ms", the log is picked up for compaction. Otherwise, Kafka uses "min.cleanable.dirty.ratio" and "min.compaction.lag.ms" to determine the log's eligibility for compaction as it does today.  
4. Add two Metrics (as described in the next section)
   
   
   

...

• Adding topic level configuration "max.compaction.lag.ms",  and corresponding broker configuration "log.cleaner.max.compaction.lag.ms", which is set to MAX_LONG by default.  If both "max.compaction.lag.ms" and "min.compaction.lag.ms" are provided in topic creation, Kafka enforces "max.compaction.lag.ms" is no less than "min.compaction.lag.ms".  This A record may remain un-compacted for this max lag, after which log cleaner threads are required to pick up the corresponding log partition for compaction unless they are currently working on other log partitions. This configuration only applies to topics that have compaction enabled. 

• Add two Metrics:  
  1) kafka.log:type=LogCleaner,name=num-logs-compacted-by-max-compaction-lag
  type: gauge
  value: the number of logs which needs to be immediately compacted as determined by max.compaction.lag in the last cleaner run from each cleaner thread.

  2) kafka.log:type=LogCleaner,name=max-compaction-delay
  type: gauge
  value: the maximum value of "compaction_finish_time - earliest_timestamp_of_first_uncompacted_segment - max.compaction.lag.ms" in the last cleaner run from each cleaner thread.
  This metric tells the delay between the time when a log is required to be picked up for compaction and the time when the compaction is done for the log. 
  

  
  

...

• By default "max.compaction.lag.ms" is set to MAX_LONG and this max compaction lag rule will not lead to additional log compaction.  There are no compatibility issues and no migration is required. 

Performance impact

• Kafka already collects compaction metrics (CleanerStats) that include how many bytes that are read/written during each compaction run and how long does it take to compact a log partition. We can also determine the frequency of log compaction by looking at how frequent compaction metrics are recorded. Those metrics can be used to measure the performance impact when adapting this KIP.  In addition, if a log partition If a log partition already gets compacted once per day before this KIP,  setting the log compaction time interval to more than one day should have little impact on the amount of resource spent on compaction since the existing log compaction configuration (e.g., min dirty ratio) will trigger log compaction before "max.compaction.lag.ms".   The added metric "num-logs-compacted-by-max-compaction-lag" can be used to determine how many log partitions are actually determined by "max.compaction.lag.ms" to be compacted.  

...

• 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.  In addition, in order to honor the max compaction lag requirement, we also need to force a roll on active segment if the required lag has passed. So the existing configuration doesn't meet requirements to ensure a maximum compaction lag.  

• Use the timestamp of the first message to estimate earliest timestamp of un-compacted segments (2-b in proposed change section) .  Since getting the timestamp of a message requires additional IOs, the estimation based on previous segment's largestTimestamp is sufficient.