...

• Enhance log compaction to support more than just offset comparison, so the insertion order isn't always dictating which records to keep (in effect, allowing for a form of OCC);
• The current behavior should remain as the default in order to minimize impact on already existing clients and avoid any migration efforts;
• Add new Kafka configuration "log.cleaner.compaction.strategy" to toggle the compaction strategy to this approach;
• Add new Topic configuration "compaction.strategy" representing the same as above;
• The default value of these configurations should be "offset", which represents the previous behavior;
• Specifically changing this to anything other than "offset" will cause the record headers to be scanned for a key matching this value. If this header is found, and its value is cast-able to a "long", then this value will be used to determine which record to keep, in a 'keep-highest' approach;
• When both records being compared contain a matching "compaction value", then the record with the highest offset will be kept;

• When both records being compared do not have a "compaction value" at all, then the record with the highest offset will be kept;

• When only one of the records being compared has a "compaction value", then this record is kept (as the other is considered to be anomalous);

...

A) This is accomplished by first converting the byte[] into an UTF-8 encoded String, and then converting this String to a long directly to long via the ByteUtils utility class. If any of these steps fails, then the compaction value is ignored (ie, its considered the record doesn't have a versioned compaction value).

Q) Why convert from byte[] to String to long instead of just converting directly to long?

A) This is done to keep the Kafka API language agnostic, as the binary representation of String is guaranteed by the used encoding, while the binary representation of long varies according to the client's coding language.

Q) Why not compare the raw byte[] directly instead of performing these conversions?

A) Although the byte[] can be compared, it is not actually comparable. As in, the binary representation of "7" is not necessarily greater than the binary representation of "5" (used for example purposes only, for low numbers these comparisons are actually correct, but they start getting into grey territory once you get some larger numbers)In order to compare these, some conversion operations would have to be done that, in the end, would not make it more efficient than directly converting to long.

Compatibility, Deprecation, and Migration Plan

...