Page History

// Producer-Recoverable
public class AbortableTransactionException extends ApiException {
    public AbortableTransactionException(String message) {
        super(message);
    }
    ...
}

//Producer-Retriable 
public class ProducerRetriableTransactionException extends ApiException {
    public ProducerRetriableTransactionException(String message) {
        super(message);
    }
 	...
}

//Producer-Retriable
public class ProducerImmediateRetriableTransactionException extends ApiException {
    public ProducerImmediateRetriableTransactionException(String message) {
        super(message);
    }
	...
}

//Application-Recoverable
public class ApplicationRecoverableTransactionException extends ApiException {
    public ApplicationRecoverableTransactionException(String message) {
        super(message);
    }
    ...
}

// Invalid-Configuration
public class InvalidConfiguationTransactionException extends ApiException {
    public InvalidConfiguationTransactionException(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);
    }
}

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 (AbortableTransactionException e) {
                    // Abortable Exception: Handle Kafka exception by aborting transaction. AbortTransaction path never throwsproducer.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();
            }
        }

    }