...

Discussion thread: here

JIRA: here 

Vote thread: here

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

...

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.

...

Proposed Changes

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

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

//Producer Retriable 
public class ProducerRetriableTransactionException extends ApiException {
 	...
}

//Producer Refresh and Retriable
public class ProducerRefreshRetriableTransactionException extends ApiException {
	...
}

//Application-Recoverable
public class ApplicationRecoverableTransactionException extends ApiException {
    ...
}

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

Client side code example

are proposing to group exceptions into four types:

1. Producer-Retriable: mostly maps onto current retriable errors. Some of these errors may indicate that a metadata update at the producer level is required. Because of that we introduce a few 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)NOTE: The producer can handle the retry on its own and the failure is invisible to the application layer.

2. Producer-Abortable: 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 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 it is good to have a mapping, I think 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.

Exception Table 

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

Yellow Image Added → Needs modification

Green Image Added → New exceptions added as part of this KIP

Check Image Added → Exception handling match as mentioned in KIP-691

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

AbortableTransactionException

ProducerRetriableTransactionException

ProducerRefreshRetriableTransactionException

ApplicationRecoverableTransactionException

InvalidConfiguationTransactionException

InvalidTxnStateException requires special handling where producer API and Transactional API has different handling:

Public Interfaces

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

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

//Producer Retriable 
public class ProducerRetriableTransactionException extends ApiException {
 	...
}

//Producer Refresh and Retriable
public class ProducerRefreshRetriableTransactionException extends ApiException {
	...
}

//Application-Recoverable
public class ApplicationRecoverableTransactionException extends ApiException {
    ...
}

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

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

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

        boolean isRunning =  true;
        // SendContinuously offsets topoll transaction for atomic commitrecords
        while (isRunning) {
            producer.sendOffsetsToTransaction(offsetsToCommit, CONSUMER_GROUP_ID);

try {
                try {
    // Commit transaction
              // Poll records from Kafka  producer.commitTransaction();
    for a timeout of 60 seconds
            } catch (AbortableTransactionException e) {
    ConsumerRecords<String, String> records = consumer.poll(ofSeconds(60));

             // Abortable Exception: Handle Kafka exception by aborting// transaction. producer.abortTransaction() should not throw abortable exception.
Process records to generate word count map
                     producer.abortTransactionMap<String, Integer> wordCountMap = new HashMap<>();

                    for resetToLastCommittedPositions(consumer);
    (ConsumerRecord<String, String> record : records) {
            }
            String[] }words catch= (InvalidConfiguationTransactionException e) {record.value().split(" ");
                //  Fatal Error: The error is bubbled up to the application layer. The application can decide what to do
for (String word : words) {
                       closeAll();
     wordCountMap.merge(word, 1, Integer::sum);
         throw e;
            } catch (KafkaException}
 | ApplicationRecoverableTransactionException e) {
                //}

 Application Recoverable: The application must restart
              // Begin closeAll();transaction
                 initializeApplication   producer.beginTransaction();

                   }
 // Produce word count results to output }

topic
         }

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

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 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-Abortable: 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 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 it is good to have a mapping, I think 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.

Exception Table 

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

Yellow Image Removed → Needs modification

Green Image Removed → New exceptions added as part of this KIP

Check Image Removed → Exception handling match as mentioned in KIP-691

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

AbortableTransactionException

...

...

Current handling

...

New Handling

...

Producer API

...

Transaction API

...

Producer API

...

Transaction API

...

TransactionAbortableException
** note – Added via KIP-890

...

Producer Abortable

...

Producer Abortable

...

Producer Abortable

...

Producer Abortable

...

TransactionAbortableException
** note – Added via KIP-890

...

Producer Abortable

...

Producer Abortable

...

Producer Abortable

...

Producer Abortable

ProducerRetriableTransactionException

...

...

Current handling

...

New Handling

...

Producer API

...

Transaction API

...

Producer API

...

Transaction API

...

CorruptRecordException

NotEnoughReplicasAfterAppendException

NotEnoughReplicasException

TimeoutException

...

Retriable if the error is retriable. Otherwise abortable

...

Retriable if the error is retriable otherwise abortable

...

...

...

ConcurrentTransactionsException

...

Retriable

...

Retriable

...

Producer Retriable (KIP-890 may see this)

...

Producer Retriable

...

CoordinatorLoadInProgressException

...

Retriable

...

Retriable

...

Producer Retriable

...

Producer Retriable

...

OutOfOrderSequenceException

...

Sometimes retriable or abortable

...

N/A

...

...

N/A

ProducerRefreshRetriableTransactionException

...

...

Current handling

...

New Handling

...

Producer API

...

Transaction API

...

Producer API

...

Transaction API

...

UnknownTopicOrPartitionException

NotLeaderOrFollowerException

...

Retriable if the error is retriable. Otherwise abortable

...

Retriable if the error is retriable otherwise abortable

...

...

...

NotCoordinatorException

CoordinatorNotAvailableException

...

Retriable

...

Retriable

...

Refresh + Retriable

...

Refresh + Retriable

ApplicationRecoverableTransactionException

...

...

Current handling

...

New Handling

...

Producer API

...

Transaction API

...

Producer API

...

Transaction API

...

IllegalStateException

...

Abortable

...

Sometimes Fatal depending on whether application or Sender caused issue. See: kafka: KAFKA-14831: Illegal state errors should be fatal in transactional producerCLOSED

...

...

...

UnknownProducerIdException

...

Sometimes retriable (no longer returned) or abortable

...

Abortable

...

...

...

ProducerFencedException

...

Fatal

...

Fatal

...

...

...

InvalidProducerEpochException

...

Abortable

...

Fatal

...

...

...

ProducerFencedException

...

Fatal

...

Fatal

...

...

...

InvalidPidMappingException

...

Sometimes retriable (no longer returned) or abortable

...

Abortable

...

...

...

UnknownProducerIdException

...

Sometimes retriable (no longer returned) or abortable

...

Abortable

...

...

...

FencedInstanceIdException

CommitFailedException

UnknownMemberIdException

IllegalGenerationExceiption

...

N/A

...

Abortable (TxnOffsetCommit Only)

...

N/A

...

...

KafkaException

...

Abortable (default seems to be abortable)

...

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

...

...

...

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

...

...

           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 (AbortableTransactionException e) {
                    // Abortable Exception: Handle Kafka exception by aborting transaction. producer.abortTransaction() should not throw abortable exception.
                    producer.abortTransaction();
                    resetToLastCommittedPositions(consumer);
                }
            } catch (InvalidConfiguationTransactionException e) {
                //  Fatal Error: The error is bubbled up to the application layer. The application can decide what to do
                closeAll();
                throw e;
            } catch (KafkaException | ApplicationRecoverableTransactionException e) {
                // Application Recoverable: The application must restart
                closeAll();
                initializeApplication();
            }
        }

    }

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

...

CorrelationIdMismatchException

...

Application Recoverable

...

Application Recoverable

InvalidConfiguationTransactionException

...

...

Current handling

...

Expected Handling

...

Producer API

...

Transaction API

...

Producer API

...

Transaction API

...

AuthenticationException

...

Abortable

...

Fatal

...

...

...

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

...

...

...

InvalidRecordException

InvalidRequiredAcksException

RecordBatchTooLargeException

InvalidTopicException

...

Retriable if the error is retriable. Otherwise abortable

...

Retriable if the error is retriable otherwise abortable

...

...

...

TopicAuthorizationException

GroupAuthorizationException

...

Abortable

...

Abortable

...

...

InvalidTxnStateException requires special handling where producer API and Transactional API has different handling:

...

...

Current handling

...

Expected Handling

...

Producer API

...

Transaction API

...

Producer API

...

Transaction API

...

InvalidTxnStateException

...

Abortable

...

Fatal

...

...

Compatibility, Deprecation, and Migration Plan

...