[This KIP proposal is a joint work between Jason Gustafson, Flavio Junqueira,  Apurva Mehta, Sriram, and guozhang Wang]

Status

Current state: Adopted

Discussion thread: http://search-hadoop.com/m/Kafka/uyzND1jwZrr7HRHf?subj=+DISCUSS+KIP+98+Exactly+Once+Delivery+and+Transactional+Messaging

JIRA: 

Motivation

This document outlines a proposal for strengthening the message delivery semantics of Kafka. This builds on significant work which has been done previously, specifically, here and here.

...

An Example Application

Here is an a simple application which demonstrates the use of the APIs introduced above.

public class KafkaTransactionsExample {
 
  public static void main(String args[]) {
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerConfig);


    // Note that the ‘transactional.id’ configuration _must_ be specified in the 
    // producer config in order to use transactions.
    KafkaProducer<String, String> producer = new KafkaProducer<>(producerConfig);

    // We need to initialize transactions once per producer instance. To use transactions, 
    // it is assumed that the application id is specified in the config with the key 
    // transactional.id.
    // 
    // This method will recover or abort transactions initiated by previous instances of a 
    // producer with the same app id. Any other transactional messages will report an error 
    // if initialization was not performed.
    //
    // The response indicates success or failure. Some failures are irrecoverable and will 
    // require a new producer  instance. See the documentation for TransactionMetadata for a 
    // list of error codes.
    producer.initTransactions();
    
    while(true) {
      ConsumerRecords<String, String> records = consumer.poll(CONSUMER_POLL_TIMEOUT);
      if (!records.isEmpty()) {
        // Start a new transaction. This will begin the process of batching the consumed 
        // records as well
        // as an records produced as a result of processing the input records. 
        //
        // We need to check the response to make sure that this producer is able to initiate 
        // a new transaction.
        producer.beginTransaction();
        
        // Process the input records and send them to the output topic(s).
        List<ProducerRecord<String, String>> outputRecords = processRecords(records);
        for (ProducerRecord<String, String> outputRecord : outputRecords) {
          producer.send(outputRecord); 
        }
        
        // To ensure that the consumed and produced messages are batched, we need to commit 
        // the offsets through
        // the producer and not the consumer.
        //
        // If this returns an error, we should abort the transaction.
        
        sendOffsetsResult = producer.sendOffsetsToTransaction(getUncommittedOffsets());
        
     
        // Now that we have consumed, processed, and produced a batch of messages, let's 
        // commit the results.
        // If this does not report success, then the transaction will be rolled back.
        producer.endTransactioncommitTransaction();
      }
    } 
  }
}

...

When a producer is finished with a transaction, the newly introduced KafkaProducer.endTransaction commitTransaction or KafkaProducer.abortTransaction must be called. The former makes the data produced in 4 available to downstream consumers. The latter effectively erases the produced data from the log: it will never be accessible to the user, ie. downstream consumers will read and discard the aborted messages.

...

// FetchRequest v4

FetchRequest => ReplicaId MaxWaitTime MinBytes IsolationLevel [TopicName [Partition FetchOffset MaxBytes]]
 ReplicaId => int32
 MaxWaitTime => int32
 MinBytes => int32
 IsolationLevel => int8 (READ_COMMITTED | READ_UNCOMMITTED)
 TopicName => string
 Partition => int32
 FetchOffset => int64
 MaxBytes => int32
 IsolationLevel => int8 (READ_COMMITTED | READ_UNCOMMITTED)

// FetchResponse v4

FetchResponse => ThrottleTime [TopicName [Partition ErrorCode HighwaterMarkOffset LastStableOffset AbortedTransactions MessageSetSize MessageSet]]
 ThrottleTime => int32
 TopicName => string
 Partition => int32
 ErrorCode => int16
 HighwaterMarkOffset => int64
 LastStableOffset => int64
 AbortedTransactions => [PID FirstOffset]
   PID => int64
   FirstOffset => int64
 MessageSetSize => int32

...

// ProduceRequest v3

ProduceRequest => TransactionalId 
                  RequiredAcks
                  Timeout
                  [TopicName [Partition MessageSetSize MessageSet]]
 TransactionalId => nullableString
 RequiredAcks => int16
 Timeout => int32
 Partition => int32
 MessageSetSize => int32
 MessageSet => bytes

...

• DuplicateSequenceNumber [NEW]

• InvalidSequenceNumber [NEW]

• ProducerFenced InvalidProducerEpoch [NEW]

• UNSUPPORTED_FOR_MESSAGE_FORMAT

...

• Ok

• NotCoordinatorForTransactionalId

• CoordinatorNotAvailable

• ConcurrentTransactions

• InvalidTransactionTimeout

...

AddPartitionsToTxnRequest => TransactionalId PID Epoch [Topic [Partition]]
 TransactionalId => string
 PID => int64
 Epoch => int32int16
 Topic => string
 Partition => int32

...

AddPartitionsToTxnResponse => ErrorCode
 ErrorCode: int16

Error code:

• Ok

• NotCoordinator

• CoordinatorNotAvailable

• ProducerFencedCoordinatorLoadInProgress

• InvalidPidMapping

• NotCoordinatorForTransactionalIdInvalidTxnState

• CoordinatorNotAvailableConcurrentTransactions

• InvalidTxnRequestGroupAuthorizationFailed

AddOffsetsToTxnRequest

Sent by the producer to its transaction coordinator to indicate a consumer offset commit operation is called as part of the current ongoing transaction. Request handling details can be found here.

AddOffsetsToTxnRequest => TransactionalId PID Epoch ConsumerGroupID
 TransactionalId => string
 PID => int64
 Epoch => int32int16
 ConsumerGroupID => string

...

AddOffsetsToTxnResponse => ErrorCode
 ErrorCode: int16
 Coordinator => NodeId Host Port
   NodeId => int32
   Host => string
   Port => int32

Error code:

Error code:

• Ok

• InvalidProducerEpoch

• InvalidPidMapping

• NotCoordinatorForTransactionalId

• CoordinatorNotAvailable

• ConcurrentTransactions

• Ok

• ProducerFenced

• InvalidPidMapping

• NotCoordinatorForTransactionalId

• CoordinatorNotAvailable

• InvalidTxnRequest

EndTxnRequest/Response

...

EndTxnRequest => TransactionalId PID Epoch Command
 TransactionalId => string
 PID => int64
 Epoch => int32int16
 Command => boolean (false(0) means ABORT, true(1) means COMMIT)

EndTxnResponse => ErrorCode
 ErrorCode => int16

Error code:

• Ok

• ProducerFencedInvalidProducerEpoch

• InvalidPidMapping

• CoordinatorNotAvailable

• ConcurrentTransactions

• NotCoordinatorForTransactionalId

• InvalidTxnRequest

...

WriteTxnMarkersRequest/Response

Sent by transaction coordinator to broker to commit the transaction. Request handling details can be found here.

WriteTxnMarkerRequestWriteTxnMarkersRequest => [CoorinadorEpoch PID Epoch Marker CoordinatorEpoch [Topic [Partition]]]
 PIDCoordinatorEpoch => int64int32
 EpochPID => int32int64
 CoordinatorEpochEpoch => int32int16
 Marker => int8boolean (false(0) =means COMMITABORT, true(1) =means ABORTCOMMIT)
 Topic => string
 Partition => int32

...

WriteTxnMarkerResponseWriteTxnMarkersResponse => ErrorCode
 ErrorCode =[PID [TopicName [Partition ErrorCode]]]
 PID => int64
 TopicName => string
 Partition => int32
 ErrorCode => int16

Error code:

• Ok

TxnOffsetCommitRequest/Response

...

TxnOffsetCommitRequest   => ConsumerGroupID
                           PID
                           Epoch
                           RetentionTime
                           OffsetAndMetadata
 ConsumerGroupID => string
 PID => int64
 Epoch => int32int16
 RetentionTime => int64
 OffsetAndMetadata => [TopicName [Partition Offset Metadata]]
   TopicName => string
   Partition => int32
   Offset => int64
   Metadata => string

...

TxnOffsetCommitResponse => [TopicName [Partition ErrorCode]]]
 TopicName => string
 Partition => int32
 ErrorCode => int16

Error code:

...

• InvalidProducerEpoch

Note: The following is tangential to the TxnOffsetCommitRequest/Response: When an  OffsetCommitRequest from a consumer failed with a retriable error, we return RetriableOffsetCommitException to the application callback. Previously, this 'RetriableOffsetCommitException' would include the underlying exception. With the changes in KIP-98, we no longer include the underlying exception in the 'RetriableOffsetCommitException'.

Message Format

In order to add new fields such as PID and epoch into the produced messages for transactional messaging and de-duplication, we need to change Kafka’s message format and bump up its version (i.e. the “magic byte”). More specifically, we need to add the following fields into each message:

...

MessageSet => 
  FirstOffset => int64
  Length => int32
  CRCPartitionLeaderEpoch => int32
 /* Magic Added for KIP-101 */
  Magic => int8  /* bump up to “2” */
  CRC => int32 /* CRC32C which covers everything from Attributes on */
  Attributes => int16
  LastOffsetDelta => int32 {NEW}
  FirstTimestamp => int64 {NEW}
  MaxTimestamp => int64 {NEW}
  PID => int64 {NEW}
  EpochProducerEpoch => int16 {NEW}
  FirstSequence => int32 {NEW}
  Messages => Message1, Message2, … , MessageN {NEW}[Message]

Message => {ALL FIELDS NEW}
  Length => uintVarvarint
  Attributes => int8
  TimestampDelta => intVarvarint
  OffsetDelta => uintVarvarint
  KeyLen => uintVar [OPTIONAL]varint
  Key => data [OPTIONAL]
  ValueLen => varint
  Value => data

  Headers => [OPTIONAL]

Header] /* See KIP-82. Note the array uses a varint for the number of headers. */
 
Header => HeaderKey HeaderVal
  HeaderKeyLen => varint
  HeaderKey => string
  HeaderValueLen => varint
  HeaderValue => data

The ability to store The ability to store some fields only at the message set level allows us to conserve space considerably when batching messages into a message set. For example, there is no need to write the PID within each message since it will always be the same for all messages within each message set. In addition, by separating the message level format and message set format, now we can also use variable-length types for the inner (relative) offsets and save considerably over a fixed 8-byte field size.

...

Message Set Attributes: The message set attributes are essentially the same as in the existing format, though we have added an additional byte for future use. In addition to the existing 3 bits used to indicate the compression codec  and 1 bit for timestamp type, we will use another bit to indicate that the message set is transactional (see Transaction Markers section). This lets consumers in READ_COMMITTED know whether a transaction marker is expected for a given message set.

Discussion Discussion on Maximum Message Size. The broker’s configuration max.message.size previously controlled the maximum size of a single uncompressed message or a compressed set of messages. With this design, it now controls the maximum message set size, compressed or not. In practice, the difference is minor because a single message can be written as a singleton message set, with the small increase in overhead mentioned above.

Message Fields

The length field of the message format is encoded as an unsigned variable-length int, abbr. “uintVar”. Similarly the offset delta and key length fields are encoded as unitVar as well. The message’s offset can then be calculated as the offset of the message set + offset delta. 

Message Attributes: In this format, we have also added a single byte for individual message attributes. Only message sets can be compressed, so there is no need to reserve some of these attributes for the compression type. Instead, we use the first bit to indicate a null key and the second bit to indicate a null value (1 indicates null, 0 indicates non-null). This allows us to use an unsigned variable length encoding for the key and value length fields. When the key or value is null, we can omit the corresponding fields:

• Null-key bit is 1: skip the key-length and key fields.

• Null-value bit is 1: skip the key-length (since it can now be calculated) and value fields.

• Both bits are 1: skip key-length, key and value fields.

The control flag indicates that the message is a control message, which means it is not intended for application consumption. The remaining bits are currently unused, though one could be used for KIP-87 (message tombstone marker).

The length field of the message format is encoded as a signed variable-length integer. Similarly the offset delta and key length fields are encoded as unitVar as well. The message’s offset can then be calculated as the offset of the message set + offset delta. 

Message Attributes: In this format, we have also added a single byte for individual message attributes. Only message sets can be compressed, so there is no need to reserve some of these attributes for the compression type. All of the message-level attributes are available for future use. 

Unused (8)

Control Messages

We use control messages to represent transaction markers. All messages contained in a batch with the control attribute set (see above) are considered control messages and follow a specific format. Each control message must

 

Null Key (1)

Null Value (1)

Control Flag (1)

Unused (5)

Control messages will always have a non-null key, which is used to indicate the type of control message type with the following schema:

...

1. Having easy access to the offset of the first message allows us to stream messages to the user on demand. In the existing format, we only know the last offset in each message set, so we have to read the messages fully into memory in order to compute the offset of the first message to be returned to the user.

2. As before, the message set header has a fixed size. This is important because it allows us to do in-place offset/timestamp assignment on the broker before writing to disk.

3. We have removed the per-message CRC in this format. We hesitated initially to do so because of its use in some auditing applications for end-to-end validation. The problem is that it is not safe, even currently, to assume that the CRC seen by the producer will match that seen by the consumer. One case where it is not preserved is when the topic is configured to use the log append time. Another is when messages need to be up-converted prior to appending to the log. For these reasons, and to conserve space and save computation, we have removed the CRC and deprecated client usage of these fields.

4. The message set CRC covers the header and message data. Alternatively, we could let it cover only the header, but if compressed data is corrupted, then decompression may fail with obscure errors. Additionally, that would require us to add the message-level CRC back to the message.

5. The CRC32C polynomial is used for all CRC computations in the new format because optimised implementations are significantly faster  (i.e. if they use the CRC32 instruction introduced in SSE4.2).

6. Individual messages within a message set have their full size (including header, key, and value) as the first field. This is designed to make deserialization efficient. As we do for the message set itself, we can read the size from the input stream, allocate memory accordingly, and do a single read up to the end of the message. This also makes it easier to skip over the messages if we are looking for a particular one, which potentially saves us from copying the key and value.

7. We have not included a field for the size of the value in the message schema since it can be computed directly using the message size and the length of the header and key.

8. We have used a variable length integer to represent timestamps. Our approach is to let the first message

...

As the batch size increases, the overhead of the new format grows smaller compared to the old format because of the eliminated redundancy. The overhead per message in the old format is fixed at 34 bytes. For the new format, the message set overhead is 53 bytes, while per-message overhead ranges from 6 to 25 bytes. This makes it more costly to send individual messages, but space is quickly recovered with even modest batching. 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) :

 

Batch Size	Old Format Overhead	New Format Overhead
1	34*1 = 34	53 + 1*7 = 60
3	34*3 = 102	53 + 3*7 = 74
10	34*10 = 340	53 + 10*7 = 123
50	34*50 = 1700	53 + 50*7 = 403
100	34

*100 = 3400

45 + 100*7 = 745

 

*100 = 3400

53 + 100*7 = 753

Metrics

As part of this work, we would need to expose new metrics to make the system operable. These would include:

1. Number of live PIDs (a proxy for the size of the PID->Sequence map)
2. Current LSO per partition (useful to detect stuck consumers and lost commit/abort markers).
3. Number of active transactionalIds (proxy for the memory consumed by the transaction coordinator).

Compatibility, Deprecation, and Migration Plan

...

...