...
Table of Contents |
---|
Status
Current state: Under Discussion Adopted
Discussion thread: here
JIRA: KAFKA-4208
...
Further details and a more detailed case for headers can be seen here : A Case for Kafka Headers
Amendments made during implementation, and on KIP-118 being pulled are highlighted orange, changes reviewed during PR and notification sent to dev mailing lists.
Public Interfaces
This KIP has the following public interface changes:
- Add a new headers length and value (byte[]) to the core message format.
Add a headers (Map<int, byte[]>) field to ProducerRecord and ConsumerRecord.
- Add accessor methods on the Producer/ConsumerRecord void setHeader(int, byte[]) and a byte[] getHeader(int)
- Add ProduceRequest/ProduceResponse V4 which uses the new message format.
- Add FetchRequest/FetchResponse V4 which uses the new message format.
- The serialisation of the [int, bye[]] header set will on the wire using a strict format
- Each headers value will be custom serialisable by the interceptors/plugins that use the header.
- As int key based headers for compactness ranges will be reserved for different usages:
- client headers adopted by the kafka open source (not necessary the kafka open source owns it)
- broker headers adopted by the kafka open source (not necessary the kafka open source owns it)
- commercial vendor space
- custom inhouse
For more detail information of the above changes, please refer to the Proposed Changes section.
Proposed Changes
There are four options proposed before this proposal. This details our proposed solution of Option 1 described here. The other options are in the Rejected Alternatives section.
- Adds the ability for producers to set standard header key=value value pairs
- No incompatible client api change (only new methods)
- Allows users to specify the serialisation of the header value per header
- Compact key space
- Provides a standardised interface to eco systems of tools that then can grow around the feature
The disadvantage of this proposal is:
- Change to the message object
Add a headers field Map<int, byte[]> to both ProducerRecord and ConsumerRecord
- Accessor methods of void setHeader(int, byte[]) and byte[] getHeader(int) added to interface of the ProducerRecord/ConsumerRecord.
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.
...
- Create a Header Interface and implementing class
Interface
Code Block public interface Header { String key(); byte[] value(); }
Implementation Detail
- Add a String key field to Header implementing class
Add a byte[] value field to Header implementing class
- Create a Headers Interface and implementing class
- Headers will be mutable
- For the Producer, after send and post interceptors it will be turned into a read only immutable instance.
- This will be done by the invoking "close()" method, this method is not exposed in the api, but an implementation detail.
Interface
Code Block public interface Headers extends Iterable<Header> { /** * Adds a header (key inside), returning if the operation succeeded. * If headers is in read-only, will always fail the operation with throwing IllegalStateException. */ Headers add(Header header) throws IllegalStateException; /** * Adds a header by key and value, returning if the operation succeeded. * If headers is in read-only, will always fail the operation with throwing IllegalStateException. */ Headers add(String key, byte[] value) throws IllegalStateException; /** * Removes ALL HEADERS for the given key returning if the operation succeeded. * If headers is in read-only, will always fail the operation with throwing IllegalStateException. */ Headers remove(String key) throws IllegalStateException; /** * Returns JUST ONE (the very last) header for the given key, if present. */ Header lastHeader(String key) /** * Returns ALL headers for the given key, if present. */ Iterable<Header> headers(String key); }
Implementation Detail
Add accessor methods on the Headers class
- Headers add(Header header)
- Headers add(String key, byte[] value)
- Headers remove(Header header)
- Header lastHeader(String key)
- Iterable<Header> headers(String key)
- Inline with ConsumerRecords interface returning the subset of ConsumerRecord for a given topic.
- public Iterable<ConsumerRecord<K, V>> records(String topic)
- Inline with ConsumerRecords interface returning the subset of ConsumerRecord for a given topic.
- implement Iterable<Header>
- interceptors and k,v serialisers are expected to add headers during the produce intercept stage.
- Headers will be mutable
- Add a headers field to ProducerRecord and ConsumerRecord.
- Add constructor(s) of Producer/ConsumerRecord to allow passing in of Iterable<Header>
- use case is MirrorMakers able to copy headers.
- Add accessor methods on the Producer/ConsumerRecord Headers headers()
Code Block public class ProducerRecord<K, V> { ... ProducerRecord(K key, V value, Iterable<Header> headers, ...) ... public Headers headers(); ... }
Code Block public class ConsumerRecord<K, V> { ... ConsumerRecord(K key, V value, Iterable<Header> headers, ...) ... public Headers headers(); ... }
- Changes needed, will piggyback onto V3 of ProduceRequest and V5 of FetchRequest which were introduced in KIP-98
- The serialisation of the [String, byte[]] header array will on the wire using a strict format
- Each headers value will be custom serialisable by the interceptors/plugins/serdes that use the header.
Expose headers to De/Serializers - extended interface added, for lack of default methods available in java 8
Code Block public interface ExtendedDeserializer<T> extends Deserializer<T> { T deserialize(String topic, Headers headers, byte[] data); }
Code Block public interface ExtendedSerializer<T> extends Serializer<T> { byte[] serialize(String topic, Headers headers, T data); }
For more detail information of the above changes, please refer to the Proposed Changes section.
Proposed Changes
There are four options proposed before this proposal. This details our proposed solution of Option 1 described here. The other options are in the Rejected Alternatives section.
- Adds the ability for producers to set standard header key=value value pairs
- No incompatible client api change (only new methods)
- Allows users to specify the serialisation of the header value per header
- Provides a standardised interface to eco systems of tools that then can grow around the feature
The disadvantage of this proposal is:
- Change to the message object
Key duplication and ordering
- Duplicate headers with the same key must be supported.
- The order of headers must be retained throughout a record's end-to-end lifetime: from producer to consumer.
Create a Headers Interface and Implementation to encapsulate headers protocol.
- See above public interfaces section for sample interfaces.
Add a headers field Headers to both ProducerRecord and ConsumerRecord
- Accessor methods of Headers headers() added to interface of the ProducerRecord/ConsumerRecord.
- Add constructor(s) of Producer/ConsumerRecord to allow passing in of an existing/pre-constructed headers via Iterable<Header>
- use case is MirrorMakers able to copy headers.
Add new method to make headers accessible during de/serialization
- Add new default method to Serialization/Deserialzation class, with Header parameter
- Due to KIP-118 not being implemented, a new extended interface ExtendedSerializer ExtendedDeserializer with new extra methods is introduced instead
- Existing De/Serialization will be wrapped and work as before.
- Due to KIP-118 not being implemented, a new extended interface ExtendedSerializer ExtendedDeserializer with new extra methods is introduced instead
Wire protocol change - add array of headers to end of the message format
The below is for the core Message wire protocol change needed to fit headers into the message.
- A header 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.
This is basing off KIP-98 Message protocol proposals.
Code Block | ||
---|---|---|
| ||
Message =>
Length => varint
Attributes => int8
TimestampDelta => varlong
OffsetDelta => varint
KeyLen => varint
Key => data
ValueLen => varint
Value => data
Headers => [Header] <------------ NEW Added Array of headers
Header =>
Key => string (utf8) <------------------------------- NEW UTF8 encoded string (uses varint length)
Value => bytes <------------------------------------ NEW header value as data (uses varint 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 which doesn't support headers
- upcasting would result in a message without headers
- new consumer would simply get a message and as no headers on send would have none.
- older consumers simply would consume a message they would not be aware of any headers
- down casting would remove headers from the message as older message format doesn't have these in protocol.
- older producers simply would produce a message which doesn't support headers
- Message version migration would be handled as like in KIP-32
- Given the mutability of the headers and the fact that we close them at some point, it means that users should not resend the same record if they use headers and interceptors.
- They should either configure the producer so that automatic retries are attempted (preferred although there are some known limitations regarding records that are timed out before a send isactually attempted)
- or
- They need to create a new record while being careful about what headers they include (if they rely fully on interceptors for headers, they could just not include any headers in the new headers).
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
The concept is similar to the above proposed but int keys
- Benefits
- more compact much reduced byte size overhead only 4 bytes.
- String keys can dwarf the value in byte size.
- Disadvantages
- String keys are more common in many systems
- Requires management of the int key space, where as string keys have natural key space management.
Map<String, String> Headers added to the ConsumerRecord
Code Block | ||
---|---|---|
| ||
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 <--------------- NEW [optional] size of the byte[] of the serialised headers if headers
(optional) Headers => bytes <--------------------- NEW [optional] serialised form of the headers Map<int, byte[]>
ValueLength => 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.
- [Discuss] should headers be ordered by the int key (ascending) or not
- [Proposed] current proposal is to not enforce ordering as headers should be composable and should be in any order, and any optimization benefits are offset by the similar perfomance disadvantages.
- Benefits
- Allows faster access to lower numbered keys which can be reserved for server side set/accessed headers
- Ability to skip/not read higher numbered keys used by clients.
- Disadvantage
- Ability to add a header to existing array is more complex, as cannot simply append
- Makes advanced uses such as repeated headers would be limited by ordering
Code Block | ||
---|---|---|
| ||
Headers (bytes) => Set(Key, ValueLength, Value)
Set =>
Key => int32 <---------------------- NEW int key of the header
ValueLength => int32 <-------------- NEW size of the byte[] of the serialised header value
Value => bytes <-------------------- NEW serialised form of the header value
|
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
Local (0x001)
- 0x00100000-0x0010FFFF Local use - Infrastructure
- 0x00110000-0x0011FFFF Local use - Client
- 0x00120000-0x0012FFFF Local use - Stack/Middleware
- 0x00130000-0x0013FFFF Local use - Application
- 0x00140000-0x0014FFFF Local use - User
- 0x00150000-0x001FFFFF RESERVED
Open (0x002)
- 0x00200000-0x0020FFFF Open use - Infrastructure
- 0x00210000-0x0021FFFF Open use - Client
- 0x00220000-0x0022FFFF Open use - Stack/Middleware
- 0x00230000-0x0023FFFF Open use - Application
- 0x00240000-0x0024FFFF Open use - User
- 0x00250000-0x002FFFFF RESERVED
Testing (0x003)
- 0x00300000-0x0030FFFF Testing - Infrastructure
- 0x00310000-0x0031FFFF Testing - Client
- 0x00320000-0x0032FFFF Testing - Stack/Middleware
- 0x00330000-0x0033FFFF Testing - Application
- 0x00340000-0x0034FFFF Testing - User
- 0x00350000-0x003FFFFF RESERVED
Reserved
- 0x00400000-0xFFFFFFFF RESERVED
- A kafka header used in the kafka open source distribution could have the value of 0x00000010 (10).
- A header used internally at organisation X would fall under Local. If it was something used by infrastructure, like tracing, that could use 0x00100010 (1048592)
- A header used by an open source project somebody is working on, that they just want to put out there (without coordination) could start using a value of 0x00220010 (2228240) if it was a plugin for annotating the location on a message.
- An application that was testing whether it can use headers for a new feature it's developing could pick a header with key 0x0034010 (3407888).
- The Kafka open source number space is coordinated via the Apache Kafka opensource project. A class would list all possible headers, their numbers and their string equivalents (for output/logging/etc).
- A Local organisation is in charge of coordinating it's local number space. It would be in charge of writing a text file, a class or a service to coordinate who gets what numbers when.
- In the open internet you can grab any number in the Open space but should expect no guarantees that other people may or may not be using that number.
- When you're doing testing you can safely grab a number in the testing space and be assured that you won't collide with an official system header. It's still possible to collide with other testing headers, but no production system should depend on these headers.
Sample register that would end up having in the open source space.
key | name | description | by | url |
---|---|---|---|---|
1 | client.id | producers client id | Apache Kafka | some url to a document about it |
2 | cluster.id | cluster id of where the message first originated | Apache Kafka | some url to a document about it |
3 | correlation.id | correlation id for where a message is for mutex response from a request | Apache Kafka | some url to a document about it |
2100001 | new.relic | stores the transaction linking guid for transaction sticking by new relic | New Relic | some url to a document about it |
2100002 | appdynamics | stores the transaction linking guid for transaction stiching by app dynamics | AppDynamics | some url to a document about it |
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.
Code Block |
---|
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
key | name | description | by | notes |
---|---|---|---|---|
1114000 | app.name | ig's unique app name for the producer | IG | some internal document about it |
1050000 | charge.tag | tag to make per message chargebacks for | IG | some internal document about it |
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
- older producers simply would produce a message without 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
...
The concept is similar to the above proposed but with a few more disadvantages.
...