Status

Current state:  Under Discussion 

...

Further details and a more detailed case for headers can be seen here : A Case for Kafka Headers

Public Interfaces

This KIP has the following public interface changes:

1. Add a new headers length and value (byte[]) to the core message format.

2. Create a Headers class 

   1. Add a headers (MultiMap<String, Object>) field to Headers class

   2. Add accessor methods on the Headers class - void put<T>(String) and a Collection<Object> get(String)

3. Add a headers Map<int, byte[]>) field to ProducerRecord and ConsumerRecord. 

4. Add accessor methods on the Producer/ConsumerRecord void setHeader(int, byte[]) and a byte[] getHeader(int)Headers getHeaders()
   1. Add ProduceRequest
   Add ProduceRequest
   1. /ProduceResponse V4 which uses the new message format.
5. Add FetchRequest/FetchResponse V4 which uses the new message format.
6. The serialisation of the [intString, byebyte[]] header set array will on the wire using a strict format
7. Each headers value will be custom serialisable by the interceptors/plugins that use the header.

For more detail information of the above changes, please refer to the Proposed Changes section.

...

Wire protocol change - use attribute bit4 as flag for if headers present. and add (optional) header size and field to the message format

The below is for the core Message wire protocol change needed to fit headers into the message.

A key to this is to ensure the headers cause no overhead if not present.

...

Wire protocol change - use attribute bit4 as flag for if headers present. and add (optional) header size and field to the message format

The below is for the core Message wire protocol change needed to fit headers into the message.

A key to this is to ensure the headers cause no overhead if not present.

• The attributes flag bit is used to keep the message size the same as before if no headers are used
  
  

MessageAndOffset => Offset MessageSize Message
  Offset => int64  
  MessageSize => int32
  
  Message => Crc MagicByte Attributes Timestamp KeyLength Key HeadersLength Headers ValueLength Value
    Crc => int32
    MagicByte => int8  <-----

MessageAndOffset => Offset MessageSize Message
  Offset => int64  
  MessageSize => int32
  
  Message => Crc MagicByte Attributes Timestamp KeyLength Key HeadersLength Headers ValueLength Value
    Crc => int32
    MagicByte => int8  <---------------------- Bump up magic byte to 2
    Attributes => int8 <---------------------- Use Bit 4 as boolean flag for if headers present
    Timestamp => int64
    KeyLength => int32
    Key => bytes
    (optional) HeadersLength => int32 <----------------- NEWBump [optional] size of the byte[] of the serialised headers if headersup magic byte to 2
    (optional) HeadersAttributes => bytesint8 <---------------------- Use Bit NEW4 [optional]as serialisedboolean formflag offor theif headers present
 Map<int, byte[]>   Timestamp => int64
    KeyLength ValueLength => int32
    ValueKey => bytes

Wire protocol of the headers bytes (if above mentioned attributes bit flag is true)

The below is for the headers wire protocol.

Headers (bytes) => Set(Key, ValueLength, Value)
  Set =>
	Key => byte [] => Typed <----    (optional) HeadersLength => int32 <----------------- NEW int[optional] keysize of the header
byte[] of the serialised Valueheaders => byte[]if headers
    (optional) Headers => Typedbytes <--------------------- NEW [optional] serialised form of the headers header value

Typed (bytes)Map<int, byte[]>
    ValueLength => (Type,int32
    Value => bytes

Wire protocol of the headers bytes (if above mentioned attributes bit flag is true)

The below is for the headers wire protocol.

• A key can occur multiple times, clients should expect to handle this, this can be represented as a multi map, or an array of values per key in clients.
  
  

Headers (bytes) => Array(KeyLength, Key, ValueLength, Value)
  Set =>
	KeyLength => int32 <-----------------NEW size of the byte[] of the serialised key value
	Key => bytes <---------------------- NEW serialised string (UTF8) bytes of the header key
    ValueLength => int32 <-------------- NEW size of the byte[] of the serialised header value
    Value => bytes <-------------------- NEW serialised form of the typed header value -> TypeValue interpretation of this below.
 
TypedValue (bytes) => (Type, Value)
	Type => byte <---------------------- 0x00 = boolean)
	Type => byte <---------------------- 0x00 = null
										 0x01 = boolean=true
                                         0x02 = boolean=false
                                         0x03 = byte
                                         0x040x01 = charbyte
                                         0x050x02 = short (int16)char
                                         0x060x03 = intshort (int32int16)
                                         0x070x04 = longint (int64int32)
                                         0x080x05 = floatlong (int64)
                                         0x090x06 = doublefloat
                                         0x0A0x07 = stringdouble
                                         0x0B0x08 = byte[]string
      Value => bytes <-------------------- NEW byte array holding the corresponding value in byte array form, 
										 0x00-0x002 - null and boolean this will be zero length
                                         0x03-0x09 - type length, will be the length of the byte array
                                         0x0A-0x0B - string and byte this will contain the leading int32 length of the byte array.
 
 

Example Key Allocation

As mentioned above ranges of keys would will be reserved for different usages, below is an example way this could be managed.

Kafka open source  (0x000)

• 0x00000000-0x0000FFFF Kafka open source
• 0x00010000-0x000FFFFF RESERVED

Sample register that would end up having in the open source space.

To assit and help ensure ease of use and uniformity a constants class should be kept and updated with the above (similar to how java sql codes work) e.g.

package org.apache.kafka.common.config;

public class KafkaHeaderKeys
{

   public static final int CLIENT_ID_KEY = 1;
   public static final int CLUSTER_ID_KEY = 2;
   public static final int CORRELATION_ID_KEY = 3;
   
}

Sample register that would end up having local (In-house) custom per organisation

Compatibility, Deprecation, and Migration Plan

• Current client users should not be affected, this is new api methods being added to client apis
• Message version allows clients expecting older message version, would not be impacted 
  • older producers simply would produce a message without headers
    • new consumer would simply get a message with empty headers
  • older consumers simply would consume a message oblivious to there being any headers
• Message version migration would be handled as like in KIP-32

Out of Scope

Some additional features/benefits were noticed and discussed on the above but are deemed out of scope and should be tackled by further KIPS.

• Core message size reduction
  • remove overhead of 4 bytes for KeyLength when no headers using attributes bit
  • reduce overhead of 4 bytes for KeyLength by using variable length encoded int
  • reduce overhead of 4 bytes for ValueLength by using variable length encoded int
• Broker side interceptors
  
  • with headers we could start introducing broker side message interceptors to append meta data or handle messages
• Single Record consumer API
  • There is many uses cases where single record consumer/listener api is more user friendly - this is evident by the fact spring kafka have already created a wrapper, it would be good to support this natively.

Rejected Alternatives

...

                                   0x09 = byte[]
    Value => bytes <-------------------- NEW byte array holding the corresponding value in byte array form, 
										 0x00-0x002 - null and boolean this will be zero length
                                         

Compatibility, Deprecation, and Migration Plan

• Current client users should not be affected, this is new api methods being added to client apis
• Message version allows clients expecting older message version, would not be impacted 
  • older producers simply would produce a message without headers
    • new consumer would simply get a message with empty headers
  • older consumers simply would consume a message oblivious to there being any headers
• Message version migration would be handled as like in KIP-32

Out of Scope

Some additional features/benefits were noticed and discussed on the above but are deemed out of scope and should be tackled by further KIPS.

• Core message size reduction
  • remove overhead of 4 bytes for KeyLength when no headers using attributes bit
  • reduce overhead of 4 bytes for KeyLength by using variable length encoded int
  • reduce overhead of 4 bytes for ValueLength by using variable length encoded int
• Broker side interceptors
  
  • with headers we could start introducing broker side message interceptors to append meta data or handle messages
• Single Record consumer API
  • There is many uses cases where single record consumer/listener api is more user friendly - this is evident by the fact spring kafka have already created a wrapper, it would be good to support this natively.

Rejected Alternatives

Map<Int, byte[]> Headers added to the Producer/ConsumerRecord

Map<String, String> Headers added to the ConsumerRecord

The concept is similar to the above proposed but with a few more disadvantages.

• Benefits

...