...

• 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. Kafka validates "max.compaction.lag.ms" is no less than "min.compaction.lag.ms".  A record may remain un-compacted for this max lag, after which the corresponding log partition becomes eligible for log compaction. This configuration only applies to topics that have compaction enabled. 

• Add the following metric:  

  1) kafka.log:type=LogCleaner,name=max-compaction-delay-secs
  type: gauge
  value: Math.max(now - earliest_timestamp_of_uncompacted_segment - max.compaction.lag.ms, 0)/10001000 
  This metric tells the max compaction delay after the time when a log is required for compaction as determined based on value is calculated across all compact-able partitions, where the max.compaction.lag.ms and the time when the compaction is done for the log.  can be overridden on per-topic basis.

Compatibility, Deprecation, and Migration Plan

...