Status

Current state:  Under Discussion

Discussion thread: : https://www.mail-archive.com/dev@kafka.apache.org/msg100621.html

Voting Thread: https://www.mail-archive.com/dev@kafka.apache.org/msg100812.htmlhere [Change the link from the KIP proposal email archive to your own email thread]

JIRA:  KAFKA-8830

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

...

Currently the RecordMetadata contains details like topic, partition, offset, timestamp etc. But there is no information on for which record this RecordMetadata belongs to. Having headers information as part of RecordMetaData will bridge this gap.

It will make the  context of the record available to interceptors Interceptors and open up possibility for building more features.

...

As the Kafka usage is growing, we see Kafka being used for Payment transactional, Settlement  and reconciliation use cases. These kind of use cases are very sensitive to message loss.Current dataloss auditing systems like chaperone uses time window based aggregations, which is not sufficient when we need message level traceability. In order to build message level traceability, we need the capability to track message production and consumption.If we choose to track message production status based on acknowledgment the producer receives, we will need ability to relate acknowledgment to the record. Having this ability in Interceptors can help us to link the record to its acknowledgement. 

Public Interfaces

We will change RecordMetaData and FutureRecordMetaData to have headers information.

...

1. Changes to RecordMetaData

   Code Block
   language java title RecordMetaData.java
   public final class RecordMetadata { /** * Partition value for record without partition assigned */ public static final int UNKNOWN_PARTITION = -1; private final long offset; .......... private final TopicPartition topicPartition; private final Header[] headers; private volatile Long checksum; public RecordMetadata(TopicPartition topicPartition, long baseOffset, long relativeOffset, long timestamp, Long checksum, int serializedKeySize, int serializedValueSize, Header[] headers) { // ignore the relativeOffset if the base offset is -1, // since this indicates the offset is unknown this.offset = baseOffset == -1 ? baseOffset : baseOffset + relativeOffset; this.timestamp = timestamp; this.checksum = checksum; this.serializedKeySize = serializedKeySize; this.serializedValueSize = serializedValueSize; this.topicPartition = topicPartition; this.headers = headers; } ...... /** * The headers of the record was sent to */ public Header[] headers() { return this.headers; } ...... }

   
   

2. Changes to FutureRecordMetaData

   Code Block
   language java title FutureRecordMetaData.java
   public final class FutureRecordMetadata implements Future<RecordMetadata> { private final ProduceRequestResult result; .......... private final Header[] headers; private volatile FutureRecordMetadata nextRecordMetadata = null; public FutureRecordMetadata(ProduceRequestResult result, long relativeOffset, long createTimestamp, Long checksum, int serializedKeySize, int serializedValueSize, Time time, Header[] headers) { this.result = result; this.relativeOffset = relativeOffset; this.createTimestamp = createTimestamp; this.checksum = checksum; this.serializedKeySize = serializedKeySize; this.serializedValueSize = serializedValueSize; this.time = time; this.headers = headers; } ....... }

   
   

3. In ProduceBatch add headers to FutureRecordMetaData while appending record to the batch.

   Code Block
   language java title ProducerBatch.java
   public FutureRecordMetadata tryAppend(long timestamp, byte[] key, byte[] value, Header[] headers, Callback callback, long now) { ....... ......... FutureRecordMetadata future = new FutureRecordMetadata(this.produceFuture, this.recordCount, timestamp, checksum, key == null ? -1 : key.length, value == null ? -1 : value.length, Time.SYSTEM, headers); // we have to keep every future returned to the users in case the batch needs to be // split to several new batches and resent. thunks.add(new Thunk(callback, future)); this.recordCount++; return future; } }

            

4. Add headers to RecordMetaData for send error scenarios also.

...