Status

Current state: Voting in progress

Discussion thread: https://lists.apache.org/thread.html/79aa6e50d7c737ddf83455dd8063692a535a1afa558620fe1a1496d3@<dev.kafka.apache.org>

JIRA: ---

PULL REQUEST: https://github.com/apache/kafka/pull/4822

Motivation

In order to use Kafka as the message broker within an Event Source architecture, it becomes essential that Kafka is able to reconstruct the current state of the events in a "most recent snapshot" approach.

This is where log compaction becomes an integral part of the workflow, as only the latest state is of interest. At the moment, Kafka accomplishes this by considering the insertion order (or highest offset) as a representation of the latest state.

The issue then occurs when the insertion order is not guaranteed, which causes the log compaction to keep the wrong state. This can be easily replicated when using a multi-threaded (or simply multiple) producer(s), or when sending the events asynchronously.

Public Interfaces

There are no changes to the public interfaces. 

Proposed Changes

• 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, with reserved values "offset" and "timestamp";
• The default value of these configurations should be "offset", which represents the previous behavior;
• When this configuration is set to "timestamp", then the record timestamp will be used to determine which record to keep, in a 'keep-highest' approach;
• When this configuration is set to anything other than "offset" or "timestamp", then the record headers are scanned for a key matching this value. If this header is found then this value (which must be of type "long") will be used to determine which record to keep, in a 'keep-highest' approach;
• When both records being compared contain a "compaction value" that is considered equal, and the configuration is set to either "timestamp" or using a custom header, 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 latest record 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);

Details of the change can be viewed from the pull request.

Compatibility, Deprecation, and Migration Plan

Following the proposed changes, there are no compatibility issues and no migration is required.

Rejected Alternatives

• Stream the data out of Kafka and perform Event Sourcing there
  • This would mean creating an in-house solution, which makes Kafka irrelevant in the design, and so its best left as a last-approach in case no solution is found on Kafka-side
• Guarantee insertion order on the producer
  • Not viable as keeping this logic synchronized greatly reduces the event throughput
• Check the version before sending the event to Kafka
  • Similar to the previous point, though it adds extra complexity as race-conditions may arise when attempting to compare
• Caching the record version as a byte array and perform the comparisons between records using a lexicographic byte array comparator
  • This adds greater flexibility on the client side, but allowing a variable byte array size to be used raises concerns about memory usage by the cache