Status

Current state: Under discussion  Complete 

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]
JIRA: here [Change the link from KAFKA-1 to your own ticket]

JIRA: here 

Vote thread: here

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

Motivation

Kafka Producer supports following transactional APIs

...

We have attempted error handing in accepted KIP: KIP-691: Enhance Transactional Producer Exception Handling - Apache Kafka - Apache Software Foundation which we can perhaps leverage some of the work here.

Public Interfaces

We would new error types which will be extended by existing exceptions mentioned in below table.

...

...

public class AbortableTransactionException extends ApiException {
    public AbortableTransactionException(String message) {
        super(message);
    }
}
public class ProducerRetriableTransactionException extends ApiException {
    public ProducerRecoverableTransactionException(String message) {
        super(message);
    }
}
public class ProducerImmediateRetriableTransactionException extends ApiException {
    public ProducerRecoverableTransactionException(String message) {
        super(message);
    }
}
public class ApplicationRecoverableTransactionException extends ApiException {
    public AbortableTransactionException(String message) {
        super(message);
    }
}
public class InvalidConfiguationTransactionException extends ApiException {
    public ProducerRecoverableTransactionException(String message) {
        super(message);
    }
}

// Extending exception types example
public class InvalidProducerEpochException extends AbortableTransactionException {
    private static final long serialVersionUID = 1L;
    public InvalidProducerEpochException(String message) {
        super(message);
    }
}

Proposed Changes

We are proposing to group exceptions into four types:

1. Producer-Retriable: mostly maps onto current retriable errors. The producer can handle the retry on its own and the failure is invisible to the application. Some of these errors may indicate that a metadata update at the producer level is required. Because of that we introduce a few two subclasses to the retriable error

   1. retry only (just send the request again – after some period of backoff)

   2. refresh metadata and retry (request metadata and maybe modify the request before resending)

2. Producer-RecoverableAbortable: mostly maps onto abortable errors. The error is bubbled to the application layer, and it can choose to roll back any state from the ongoing transaction and abort the transaction. The producer does not need to restart, as after aborting, it can be confident that the state was as it was before the transaction started. The application can also close the producer and react as it does for application-recoverable cases, but it doesn’t need to.

3. Application-Recoverable: maps on to some fatal errors. The error is bubbled to the application layer, and it may need to do a bit more to roll back and clean up state. The producer can not simply abort and know the current state of the partitions. Applications can handle this in different ways – streams Kafka Streams may rebalance a task and/or close the task and restart it. Another application may read from their own checkpoint to continue. In any case, the producer must restart and will be unusable after encountering this error.

4. Invalid-Configuration: maps to some fatal errors. The error is bubbled up to the application layer. The application can decide what to do. The producer doesn’t need to restart, but the application may chose to close it.

Each error code always represent the same class and rarely rely on client state to determine how to handle.While  Additionally, while it is good to have a mapping, I think it It is also useful to have a general strategy – ie a typical unknown error (not specified by the client to have a type) should probably be application recoverable.

Note: For Retriable errors, the producer handles retries internally, keeping the failure details hidden from the application. Conversely, other types of exceptions will be surfaced to the application code for handling.

Exception Table 

Below table contains list of exceptions with current and expected handingNew handing. All exceptions are grouped under a header which represent Exception Name:

Yellow → Needs modificationGreen Image Removed → New exceptions added as part of this KIP

Check → Exception handling match as mentioned in KIP-691

Cross → Exception handling doesn't match as mentioned in KIP-691

RetriableException

RefreshRetriableException

TransactionAbortableException

ApplicationRecoverableException

InvalidConfigurationException

Producer Retriable

Exception/Error Names	Current handling		Expected Handling
	Producer API	Transaction API	Producer API	Transaction API	Comments
AuthenticationException	Abortable	Fatal	Invalid Configuration Image Added	Invalid Configuration (not expected) Image Modified
ClusterAuthorizationException TransactionalIdAuthorizationException UnsupportedVersionException UnsupportedForMessageFormatException	Fatal, except UnsupportedForMessageFormatException which is Abortable	Cluster/Transaction Auth → abortable on InitProducerId, other errors abortable TransactionAuth → fatal, others abortable for AddPartitions, Find Coordinator, EndTxn, AddOffsets Transaction Auth, UnssupportedForMessageFormat → fatal for OffsetCommit	Invalid Configuration Image Modified	Invalid Configuration

Image Removed

CorruptRecordException

NotEnoughReplicasAfterAppendException

NotEnoughReplicasException

TimeoutException

Retriable if the error is retriable. Otherwise abortable

Retriable if the error is retriable otherwise abortable

Producer Retriable Image Removed

Image Modified

UnknownTopicOrPartitionException

NotLeaderOrFollowerException

Retriable if the error is retriable. Otherwise abortable

Retriable if the error is retriable otherwise abortable

Refresh + Retriable Image Removed

Refresh + Retriable Image Removed

InvalidRecordException InvalidRequiredAcksException RecordBatchTooLargeException InvalidTopicException	Retriable if the error is retriable. Otherwise abortable	Retriable if the error is retriable otherwise abortable	Invalid Configuration Image Modified	Invalid Configuration Image Modified
TopicAuthorizationException GroupAuthorizationException	Abortable	Abortable	Invalid Configuration Image Modified	Invalid Configuration Image Modified

FencedInstanceIdException

CommitFailedException

UnknownMemberIdException

IllegalGenerationExceiption

N/A

Abortable (TxnOffsetCommit Only)

N/A

Application Recoverable Image Removed

InvalidProducerEpochException

• If the PID’s epoch number is different from the current TransactionalId PID mapping, reply with the InvalidProducerEpoch error code; otherwise proceed to next step.

Abortable

Fatal

Application Recoverable Image Removed

Application Recoverable Image Removed

ProducerFencedException

Fatal

Fatal

Application Recoverable Image Removed

Application Recoverable Image Removed

OutOfOrderSequenceException

Sometimes retriable or abortable

N/A

Producer Retriable Image Removed

N/A

InvalidTxnStateException

• When client makes a request which involves change of Txn state. However expected state transition is invalid.
  e.g COMPLETE_ABORT to COMMIT

Abortable

Fatal

Producer Recoverable (KIP-890 relies on this – note this is the only one that differs with Produce API) Image Removed

Application Recoverable Image Removed

KafkaException

Abortable (default seems to be abortable)

Fatal in most cases, but abortable when there are partition errors

Application Recoverable (not expected) Image Removed

Application Recoverable (not expected) Image Removed

RuntimeException

Abortable (default seems to be abortable)

Fatal – only thrown as this generic type when correlation ID is wrong. This should be updated as KIP-691 suggests

Application Recoverable (not expected) Image Removed

Application Recoverable (not expected) Image Removed

ConcurrentTransactionsException

Retriable

Retriable

Producer Retriable (KIP-890 may see this)

Producer Retriable

CoordinatorLoadInProgressException

Retriable

Retriable

Producer Retriable

Producer Retriable

NotCoordinatorException

CoordinatorNotAvailableException

• Check whether there is a previous entry with the same TransactionalId and a higher epoch. If so, throw an exception. In particular, this indicates the log is corrupt. All future transactional RPCs to this coordintaor will result in a `NotCoordinatorForTransactionalId` error code, and this partition of the log will be effectively disabled.

• Check if it is the assigned transaction coordinator for the TransactionalId, if not reply with the NotCoordinatorForTransactionalId error code.

Retriable

Retriable

Refresh + Retriable

Refresh + Retriable

CorrelationIdMismatchException

Application Recoverable

Application Recoverable

Compatibility, Deprecation, and Migration Plan

Public Interfaces

We have below exception classes available in current Kafka code

// Producer Abortable Transaction
public class TransactionAbortableException extends ApiException {
    ...
}

//Producer Retriable  
public abstract class RetriableException extends ApiException {
   ...
}


// Invalid-Configuration
public class InvalidConfigurationException extends ApiException {
 	...
}

We will add new exception types, as listed in the table below, that extend the existing exceptions.

//Producer Refresh and Retriable
public abstract class RefreshRetriableException extends RetriableException {
	...
}

//Application-Recoverable new
public abstract class ApplicationRecoverableException extends ApiException {
    ...
}

We will extend below existing exceptions types to maintain class hierarchy 

// UnknownTopicOrPartitionException and NotLeaderOrFollowerException  extends InvalidMetadataException 
public class InvalidMetadataException extends RefreshRetriableException {
	...
}

// TopicAuthorizationException and GroupAuthorizationException extends AuthorizationException 
public class AuthorizationException extends InvalidConfigurationException {
	...
}

Client side code example

public class TransactionalClientDemo {

    private static final String CONSUMER_GROUP_ID = "my-group-id";
    private static final String OUTPUT_TOPIC = "output";
    private static final String INPUT_TOPIC = "input";
    private static KafkaConsumer<String, String> consumer;
    private static KafkaProducer<String, String> producer;

    public static void main(String[] args) {
        initializeApplication();

        boolean isRunning = true;
        // Continuously poll for records
        while (isRunning) {
            try {
                try {
                    // Poll records from Kafka for a timeout of 60 seconds
                    ConsumerRecords<String, String> records = consumer.poll(ofSeconds(60));

                    // Process records to generate word count map
                    Map<String, Integer> wordCountMap = new HashMap<>();

                    for (ConsumerRecord<String, String> record : records) {
                        String[] words = record.value().split(" ");
                        for (String word : words) {
						  wordCountMap.merge(word, 1, Integer::sum);
                        }
                    }

                    // Begin transaction
                    producer.beginTransaction();

                    // Produce word count results to output topic
                    wordCountMap.forEach((key, value) ->
                            producer.send(new ProducerRecord<>(OUTPUT_TOPIC, key, value.toString())));

                    // Determine offsets to commit
                    Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();
                    for (TopicPartition partition : records.partitions()) {
                        List<ConsumerRecord<String, String>> partitionedRecords = records.records(partition);
                        long offset = partitionedRecords.get(partitionedRecords.size() - 1).offset();
                        offsetsToCommit.put(partition, new OffsetAndMetadata(offset + 1));
                    }

                    // Send offsets to transaction for atomic commit
                    producer.sendOffsetsToTransaction(offsetsToCommit, CONSUMER_GROUP_ID);

                    // Commit transaction
                    producer.commitTransaction();
                } catch (TransactionAbortableException e) {
                    // Abortable Exception: Handle Kafka exception by aborting transaction. producer.abortTransaction() should not throw abortable exception.
                    producer.abortTransaction();
                    resetToLastCommittedPositions(consumer);
                }
            } catch (InvalidConfigurationException e) {
                //  Fatal Error: The error is bubbled up to the application layer. The application can decide what to do
                closeAll();
                throw e;
            } catch (KafkaException | ApplicationRecoverableException e) {
                // Application Recoverable: The application must restart
                closeAll();
                initializeApplication();
            }
        }

    }

Full example can be accessed at: https://github.com/apache/kafka/pull/15913/files 

Note: The producer.abortTransaction() method will not throw an abortable exception to avoid creating a loop.

Compatibility, Deprecation, and Migration Plan

For applications using older code

For existing errors, if the application code already handles the error, it will continue to do so. However, the new TransactionAbortableException  introduced in KIP-890 extends KafkaException, which is treated as fatal by applications. Therefore, we expect TransactionAbortableException  to also be treated as fatal by older code.

Currently, the transactional producer.send returns retriable exception types, such as TimeoutException , which poses a risk of duplicates in Kafka. In this KIP, we will update the transactional producer.send path such that all retriable exceptions will be translated to TransactionAbortableException. Older clients that are using the transactional producer and handling TimeoutException  by retrying the produce operation can update to handle TransactionAbortableException .

For applications using newer code

Applications using newer code, as mentioned in Clientsidecodeexample, have the flexibility to adjust their exception handling logic based on the specific type of exception encountered, enabling them to take appropriate actions.

Handling for OutOfOrderSequenceException  requires additional considerations which is out of scope for this KIP.  OutOfOrderSequenceException  and UnknownProducerIdException  (which extends OutOfOrderSequenceException  ) will not be categorised under any of the defined exception groups in this KIP. This will not impact applications using newer codeThis KIP involves client side changes which only affects the resiliency of new Producer client and Streams. Old clients will continue to be able to use their error handling. Based on type of exception thrown, user needs to change their exception catching logic to take actions against their exception handling.

Test Plan

We will add additional integration and unit tests to check if client acts as expected.

Rejected Alternatives

We discussed another approach where we encode the handling into the response as a separate field from the error code or something similar. Produce responses have space for a code and for a message, so something similar could be done to distinguish the cause/meaning of the error from the action that should be taken. The encoding of the error could be included in the definition of the response spec for other client compatibility. We rejected this approach as it requires a larger overhaul on the APIs as well as the clients.

...