...

• 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".  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. We only collect this max-compaction-delay for log compaction that is determined by max.compaction.lag. 
  

  
  

Compatibility, Deprecation, and Migration Plan

...