...
This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.
Status
Current state: "Under Discussiondiscarded"
Discussion thread: here
JIRA: here
Note: discarding this KIP since the benefits of the space savings outweigh the cost incurred for message conversion. And there are also planned changes for version v.3, like:
3. v3 of the record format should make it cheaper to make changes in the future (perhaps it could support tagged fields or similar)
4. We'd want to fix other known issues at the same time (eg log append time should always be available, there may be others)
5. We should consider whether we would want to introduce a user header that is at the batch level vs record level for efficiency reasons
[Additional proposal] Changing the batch format proposal
Problem: Currently, we perform validation of data schema (such as verifying monotonically increasing offsets etc.). To do that, we have to read the headers of each message in a batch. For a compressed batch, this means that we have to decompress the entire batch including the payload (key/value) even though we don't really require any information out of them.
Solution: If we change the ordering of messages in a batch to prefix with headers of all messages first and followed by the payload of all messages, it will lead to an optimization where we have to "partially" decompress the batch (i.e. the prefix).
Ideas for message format v.3 wiki: ideas for kafka message format v.3
Motivation
Logs in kafka consists of batches, and one batch consists of many messages. So if the size of each message header can be reduced, it'll improve the network traffic and storage size (and money, of course) a lot, even it's a small improvement. Some companies now handle trillions of messages per day using Kafka, so, if we can reduce just 1 byte per message, the save is really considerable.
...
Code Block |
---|
baseOffset: int64 batchLength: int32 partitionLeaderEpoch: int32 magic: int8 (current magic value is 2) // <-- will bump to 3 crc: int32 attributes: int16 bit 0~2: 0: no compression 1: gzip 2: snappy 3: lz4 4: zstd bit 3: timestampType bit 4: isTransactional (0 means not transactional) bit 5: isControlBatch (0 means not a control batch) bit 6: hasDeleteHorizonMs (0 means baseTimestamp is not set as the delete horizon for compaction) // new added attribute below bit 7: ignoreMessageAttributes (0 means not to ignore) bit 8~15: unused producerId: int64 producerEpoch: int16 baseSequence: int32 lastOffsetDelta: int32 // <-- will change to varint baseTimestamp: int64 // <-- will change to varlong maxTimestamp: int64 // <-- will change semantic to maxTimestampDelta and change type to varint producerId: int64 producerEpoch: int16 baseSequence: int32 records: [Record] |
Furthermore, the record batch header can also be smaller. I'd also like to improve them by:
...
3. lastOffsetDelta: change the type from int32 to varint. Same as above, In most case, the offset delta should be small. So, changing to varint will save space.
4. move above 3 fields to the trail of the batch headers.
Note:
The biggest value of varint 2 bytes is 16383
...
[8 - 6 (baseTimestamp)] + [8 - 2 (max timestamp delta)] + [4 - 2 (offset delta)] = 10 Bytes for each batch.
18% of space save for batch overhead, compared with original batch overhead 53 bytes.
For the each message header space save, using the example in KIP-98:
For example, assuming a fixed message size of 1K with 100 byte keys and reasonably close timestamps, the overhead increases by only 7 bytes for each additional batched message (2 bytes for the message size, 1 byte for attributes, 2 bytes for timestamp delta, 1 byte for offset delta, and 1 byte for key size)
14% space save for each message overhead, compared with originally 7 bytes, now only 6 bytes needed.
Proposed Changes
When writing the batch using the new version of message format, we'll default set ignoreMessageAttributes field to 1 in record batch header, and create records without attribute field.
...