[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();
      }
    } 
  }
}

In additional to the new concepts above, we also introduce new request types, new versions of existing requests, and new versions of the core message format in order to support transactions. The details of all of these will be deferred to other documents.

Data Flow

Image RemovedImage Added

In the diagram above, the sharp edged boxes represent distinct machines. The rounded boxes at the bottom represent Kafka TopicPartitions, and the diagonally rounded boxes represent logical entities which run inside brokers. 

...

If no TransactionalId is specified in the configuration, a fresh PID is assigned, and the producer only enjoys idempotent semantics and transactional semantics within a single session.

3. Starting a Transaction – The

...

beginTransaction() API

The new KafkaProducer will have a KafkaProducer.beginTxnbeginTransaction() method which has to be called to signal the start of a new transaction. Within this method, the producer will send a BeginTxnRequest to the transaction coordinator, which will record the start of the transaction in the transaction log as denoted in step 3aThe producer records local state indicating that the transaction has begun, but the transaction won’t begin from the coordinator’s perspective until the first record is sent.

4. The consume-transform-produce loop

...

The producer sends this request to the transaction coordinator the first time a new TopicPartition is written to as part of a transaction. The addition of this TopicPartition to the transaction is logged by the coordinator in step 4.1a. We need this information so that we can write the commit or abort markers to each TopicPartition (see section 5.2 for details). If this is the first partition added to the transaction, the coordinator will also start the transaction timer.

4.2 ProduceRequest

The producer writes a bunch of messages to the user’s TopicPartitions through one or more ProduceRequests (fired from the send method of the producer). These requests include the PID , epoch, and sequence number as denoted in 4.2a.

...

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.

...

The transaction coordinator handles each of the following requests: InitPid, BeginTxn, AddPartitionsToTxn, AddOffsetsToTxn, and EndTxn. Each request to the transaction coordinator includes the producer’s TransactionalId and can be used for authorization. Each of these requests mutates the transaction state of the producer, so they all require Write access to the corresponding ProducerTransactionalId resource.  Additionally, the AddPartitionsToTxn API requires Write access to the topics corresponding to the included partitions, and the AddOffsetsToTxn API requires Read access to the group included in the request.

...

// 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

...

• DuplicateSequenceNumber [NEW]

• InvalidSequenceNumber [NEW]

• ProducerFenced InvalidProducerEpoch [NEW]

• UNSUPPORTED_FOR_MESSAGE_FORMAT

...

InitPidRequest => TransactionalId TransactionTimeoutMs
 TransactionalId => String
 TransactionTimeoutMs => int32

InitPIDResponse => Error PID Epoch
 Error => Int16
 PID => Int64
 Epoch => Int16

...

• Ok

• NotCoordinatorForTransactionalId

• CoordinatorNotAvailable

...

• ConcurrentTransactions

• InvalidTransactionTimeout

AddPartitionsToTxnRequest/Response

Sent by producer to its transaction coordinator to to begin a new add a partition to the current ongoing transaction. Request handling details can be found here.

BeginTxnRequestAddPartitionsToTxnRequest => TransactionalId PID Epoch [Topic [Partition]]
 TransactionalId => string
 PID => int64
 Epoch => int16
 Topic => string
 Partition => int32

BeginTxnResponseAddPartitionsToTxnResponse => ErrorCode
 ErrorCode =>: int16

Error code:

• Ok

• NotCoordinator

• CoordinatorNotAvailable

• CoordinatorLoadInProgressProducerFenced

• InvalidPidMapping

• NotCoordinatorForTransactionalIdInvalidTxnState

• CoordinatorNotAvailableConcurrentTransactions

• InvalidTxnRequestGroupAuthorizationFailed

...

AddOffsetsToTxnRequest

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

AddPartitionsToTxnRequestAddOffsetsToTxnRequest => TransactionalId PID Epoch [Topic [Partition]]ConsumerGroupID
 TransactionalId => string
 PID => int64
 Epoch => int32int16
 TopicConsumerGroupID => string
 Partition => int32

AddPartitionsToTxnResponseAddOffsetsToTxnResponse => ErrorCode
 ErrorCode: int16

Error code:

• Ok

• ProducerFencedInvalidProducerEpoch

• InvalidPidMapping

• NotCoordinatorForTransactionalId

• CoordinatorNotAvailable

• ConcurrentTransactions

• InvalidTxnRequest

...

EndTxnRequest/Response

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

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

AddOffsetsToTxnResponseEndTxnResponse => ErrorCode
 ErrorCode: int16
 Coordinator => NodeId Host Port
   NodeId => int32
   Host => string
   Port => int32int16

Error code:

• Ok

• ProducerFencedInvalidProducerEpoch

• InvalidPidMapping

• CoordinatorNotAvailable

• ConcurrentTransactions

• NotCoordinatorForTransactionalIdCoordinatorNotAvailable

• InvalidTxnRequest

...

WriteTxnMarkersRequest/Response

Sent by producer to its transaction coordinator to prepare committing or aborting the current ongoing to broker to commit the transaction. Request handling details can be found here.

EndTxnRequestWriteTxnMarkersRequest => TransactionalId[CoorinadorEpoch PID Epoch Command
Marker TransactionalId =[Topic [Partition]]]
 CoordinatorEpoch => stringint32
 PID => int64
 Epoch => int32int16
 CommandMarker => boolean (false(0) means ABORT, true(1) means COMMIT)

EndTxnResponse
 Topic => ErrorCodestring
 ErrorCodePartition => int16int32

...

...

Ok

...

ProducerFenced

...

InvalidPidMapping

...

CoordinatorNotAvailable

...

...

InvalidTxnRequest

WriteTxnMarkerRequest/Response

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

WriteTxnMarkerRequestWriteTxnMarkersResponse => [PID Epoch Marker CoordinatorEpoch [TopicTopicName [Partition ErrorCode]]]
 PID => int64
 EpochTopicName => int32string
 CoordinatorEpochPartition => int32
 MarkerErrorCode => int8 (0 = COMMIT, 1 = ABORT)
 Topic => string
 Partition => int32

WriteTxnMarkerResponse => ErrorCode
 ErrorCode => int16

int16

Error 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:

• ProducerFenced

Message Format

• 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 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 /* Added for KIP-101 */
  Magic => int8  /* bump up to “2” */
  AttributesCRC => int16
int32 /* LastOffsetDeltaCRC32C =>which int32 {NEW}
  FirstTimestamp => int64covers everything from Attributes on */
  Attributes => int16
  LastOffsetDelta => int32 {NEW}
  FirstTimestamp => int64 {NEW}
  MaxTimestampDeltaMaxTimestamp => 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 some fields only at the 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 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.

...

The length field of the message format is encoded as an unsigned a signed variable-length int, abbr. “uintVar”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. At the end we still maintains a message-level CRC (reason discussed below). 

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).

. All of the message-level attributes are available for future use. 

...

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) :

Compatibility, Deprecation, and Migration Plan

...

...