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

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

Example: a transactional client with consistent error handling

The following example, adapted from the TransactionalClientDemo 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.

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.

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.

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

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

Exception hierarchy

Example: a transactional client with consistent error handling

Use this hierarchy in Kafka Streams applications

Related content