<a id="streams-developer-guide-handle-transactional-errors"></a>

# Handle Transactional Errors in Kafka Streams and Kafka Clients for Confluent Platform

Applications that use the transactional producer and consumer APIs to
achieve exactly-once semantics must react correctly when a
transactional operation fails.

[KIP-1050: Consistent error handling for Transactions](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1050%3A+Consistent+error+handling+for+Transactions)
standardizes this behavior by organizing every exception that a
transactional producer or consumer can raise into a small set of categories.
Each category maps to exactly one recovery action, which makes error
handling in transactional applications predictable across the Java Client.

This exception hierarchy is available starting with `kafka-clients` 4.1.0,
bundled with Confluent Platform 8.1 and later.

<a id="streams-developer-guide-handle-transactional-errors-hierarchy"></a>

## Exception hierarchy

The following table lists the exception categories introduced by KIP-1050,
in order from least to most disruptive, along with the action your
application must take for each one.

| Category                          | Description                                                                                                                                                                                                                              | Required application action                                                                                                                                                                                                  |
|-----------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `RetriableException`              | A transient error that the client retries automatically. The<br/>exception never reaches application code.                                                                                                                               | None. Handled internally by the client.                                                                                                                                                                                      |
| `RefreshRetriableException`       | A subclass of `RetriableException` for errors, such as stale<br/>partition leadership, that require the client to refresh metadata<br/>before retrying.                                                                                  | None. Handled internally by the client after a metadata refresh.                                                                                                                                                             |
| `TransactionAbortableException`   | The current transaction can’t be committed, but the producer instance<br/>itself is still usable. Originally introduced by<br/>[KIP-890](https://cwiki.apache.org/confluence/display/KAFKA/KIP-890%3A+Transactions+Server-Side+Defense). | Call `producer.abortTransaction()`, reset the consumer to its last<br/>committed positions, and reprocess the aborted records. Don’t restart<br/>the producer or consumer.                                                   |
| `ApplicationRecoverableException` | A fatal producer error, such as `ProducerFencedException` or<br/>`InvalidProducerEpochException`. The producer can no longer be used.                                                                                                    | Close and re-create the producer and the consumer, then call<br/>`initTransactions()` again before resuming processing.                                                                                                      |
| `InvalidConfigurationException`   | The transactional client is misconfigured. For example, an invalid<br/>`transactional.id` or replication setting.                                                                                                                        | Thrown directly to the calling code; the client doesn’t retry or<br/>recover from it automatically and doesn’t need to restart. Close the<br/>client and stop, because retrying with the same configuration fails<br/>again. |
| `KafkaException`                  | Any other exception that doesn’t fall into one of the preceding<br/>categories.                                                                                                                                                          | Treat as `ApplicationRecoverableException`: close and re-create the<br/>producer and consumer.                                                                                                                               |

`TimeoutException` isn’t one of the KIP-1050 categories, but a timed-out
transactional request risks message duplication if it’s retried blindly.
Handle a `TimeoutException` from a transactional operation the same way as
`TransactionAbortableException`: abort the transaction and reprocess the
records.

#### NOTE
`OutOfOrderSequenceException` and `UnknownProducerIdException` predate
KIP-1050 and aren’t assigned to one of these categories. Continue to
handle them as you did previously.

<a id="streams-developer-guide-handle-transactional-errors-example"></a>

## Example: a transactional client with consistent error handling

The following example, adapted from the
[TransactionalClientDemo](https://github.com/apache/kafka/blob/trunk/examples/src/main/java/kafka/examples/TransactionalClientDemo.java)
in the Kafka source repository, consumes records from an input topic,
computes a word count, and produces the result to an output topic inside a
transaction. The nested `try`/`catch` blocks show the recovery action for
each exception category:

- The inner `catch` handles `TransactionAbortableException` by aborting
  the transaction and rewinding the consumer, without tearing down the
  clients.
- The outer `catch` blocks handle `InvalidConfigurationException` to shut
  down, and `ApplicationRecoverableException` and `KafkaException`
  to re-create the clients and call `initTransactions()` again.

```java
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;
    private static volatile boolean isRunning = true;

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

        while (isRunning) {
            try {
                try {
                    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);
                        }
                    }

                    producer.beginTransaction();

                    wordCountMap.forEach((key, value) ->
                        producer.send(new ProducerRecord<>(OUTPUT_TOPIC, key, value.toString())));

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

                    producer.sendOffsetsToTransaction(offsetsToCommit, consumer.groupMetadata());
                    producer.commitTransaction();
                } catch (TransactionAbortableException e) {
                    // Abortable: the producer is still usable after aborting.
                    // producer.abortTransaction() should not itself throw an abortable exception.
                    producer.abortTransaction();
                    resetToLastCommittedPositions(consumer);
                }
            } catch (InvalidConfigurationException e) {
                // Fatal: thrown directly to the caller, which decides what to do.
                closeAll();
                throw e;
            } catch (ApplicationRecoverableException e) {
                // Application recoverable: the client must restart.
                closeAll();
                initializeApplication();
            } catch (KafkaException e) {
                // Treat as application recoverable unless you have a more specific policy.
                closeAll();
                initializeApplication();
            }
        }
    }

    public static void initializeApplication() {
        consumer = createKafkaConsumer();
        producer = createKafkaProducer();
        producer.initTransactions();
    }

    private static KafkaConsumer<String, String> createKafkaConsumer() {
        Properties props = new Properties();
        props.put(BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(GROUP_ID_CONFIG, CONSUMER_GROUP_ID);
        props.put(ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ISOLATION_LEVEL_CONFIG, "read_committed");
        props.put(KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(singleton(INPUT_TOPIC));
        return consumer;
    }

    private static KafkaProducer<String, String> createKafkaProducer() {
        Properties props = new Properties();
        props.put(BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ENABLE_IDEMPOTENCE_CONFIG, "true");
        props.put(TRANSACTIONAL_ID_CONFIG, "prod-1");
        props.put(KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        return new KafkaProducer<>(props);
    }

    private static void resetToLastCommittedPositions(KafkaConsumer<String, String> consumer) {
        final Map<TopicPartition, OffsetAndMetadata> committed = consumer.committed(consumer.assignment());
        consumer.assignment().forEach(tp -> {
            OffsetAndMetadata offsetAndMetadata = committed.get(tp);
            if (offsetAndMetadata != null)
                consumer.seek(tp, offsetAndMetadata.offset());
            else
                consumer.seekToBeginning(singleton(tp));
        });
    }

    private static void closeAll() {
        if (consumer != null) {
            consumer.close();
        }
        if (producer != null) {
            producer.close();
        }
    }

    private static void registerShutdownHook() {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            isRunning = false;
            closeAll();
        }));
    }
}
```

For the full, runnable version of this example, including logging and
imports, see
[TransactionalClientDemo.java](https://github.com/apache/kafka/blob/trunk/examples/src/main/java/kafka/examples/TransactionalClientDemo.java).

<a id="streams-developer-guide-handle-transactional-errors-kstreams"></a>

## Use this hierarchy in Kafka Streams applications

The preceding example uses the plain `KafkaProducer` and `KafkaConsumer`
APIs directly. When you set `processing.guarantee=exactly_once_v2` in a
Kafka Streams application, Kafka Streams manages the transactional producer and
consumer for each stream thread internally, so you don’t call
`beginTransaction()`, `commitTransaction()`, or
`sendOffsetsToTransaction()` yourself. For the related configuration, see
[processing.guarantee](config-streams.md#streams-developer-guide-processing-guarantee).

This exception hierarchy still matters directly to a Kafka Streams application
in these cases:

- You register an `UncaughtExceptionHandler` on your `KafkaStreams`
  instance, as described in
  [Write a Kafka Streams Application for Confluent Platform](write-streams.md#streams-write-app). Knowing which category an underlying exception
  belongs to helps you choose a response that’s appropriate to the failure,
  such as `REPLACE_THREAD`, `SHUTDOWN_CLIENT`, or
  `SHUTDOWN_APPLICATION`. For example, a
  `TransactionAbortableException` is generally safe to recover from with
  `REPLACE_THREAD`, while an `ApplicationRecoverableException` or
  `InvalidConfigurationException` is not.
- You use your own transactional producer or consumer alongside Kafka Streams.
  For example, to write to an external system transactionally from a custom
  processor, follow the same recovery pattern shown in the
  preceding example.

<a id="streams-developer-guide-handle-transactional-errors-more"></a>

## Related content

- [KIP-1050: Consistent error handling for Transactions](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1050%3A+Consistent+error+handling+for+Transactions)
- [KIP-890: Transactions Server-Side Defense](https://cwiki.apache.org/confluence/display/KAFKA/KIP-890%3A+Transactions+Server-Side+Defense)
- [TransactionalClientDemo.java](https://github.com/apache/kafka/blob/trunk/examples/src/main/java/kafka/examples/TransactionalClientDemo.java)
- [processing.guarantee](config-streams.md#streams-developer-guide-processing-guarantee)
- [Write a Kafka Streams Application for Confluent Platform](write-streams.md#streams-write-app)

#### NOTE
This website includes content developed at the [Apache Software Foundation](https://www.apache.org/)
under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).