@Override
    public void run() {
        int retries = 0;
        int processedRecords = 0;
        long remainingRecords = Long.MAX_VALUE;

        // it is recommended to have a relatively short txn timeout in order to clear pending offsets faster
        int transactionTimeoutMs = 10_000;
        // consumer must be in read_committed mode, which means it won't be able to read uncommitted data
        boolean readCommitted = true;

        try (KafkaProducer<Integer, String> producer = new Producer("processor-producer", bootstrapServers, outputTopic,
                true, transactionalId, true, -1, transactionTimeoutMs, null).createKafkaProducer();
             KafkaConsumer<Integer, String> consumer = new Consumer("processor-consumer", bootstrapServers, inputTopic,
                 "processor-group", Optional.of(groupInstanceId), readCommitted, -1, null).createKafkaConsumer()) {
            // called first and once to fence zombies and abort any pending transaction
            producer.initTransactions();
            consumer.subscribe(singleton(inputTopic), this);

            Utils.printOut("Processing new records");
            while (!closed && remainingRecords > 0) {
                try {
                    ConsumerRecords<Integer, String> records = consumer.poll(ofMillis(200));
                    if (!records.isEmpty()) {
                        // begin a new transaction session
                        producer.beginTransaction();

                        for (ConsumerRecord<Integer, String> record : records) {
                            // process the record and send downstream
                            ProducerRecord<Integer, String> newRecord =
                                new ProducerRecord<>(outputTopic, record.key(), record.value() + "-ok");
                            producer.send(newRecord);
                        }

                        // checkpoint the progress by sending offsets to group coordinator broker
                        // note that this API is only available for broker >= 2.5
                        producer.sendOffsetsToTransaction(getOffsetsToCommit(consumer), consumer.groupMetadata());

                        // commit the transaction including offsets
                        producer.commitTransaction();
                        processedRecords += records.count();
                        retries = 0;
                    }
                } catch (AuthorizationException | UnsupportedVersionException | ProducerFencedException
                         | FencedInstanceIdException | OutOfOrderSequenceException | SerializationException e) {
                    // we can't recover from these exceptions
                    Utils.printErr(e.getMessage());
                    shutdown();
                } catch (OffsetOutOfRangeException | NoOffsetForPartitionException e) {
                    // invalid or no offset found without auto.reset.policy
                    Utils.printOut("Invalid or no offset found, using latest");
                    consumer.seekToEnd(emptyList());
                    consumer.commitSync();
                    retries = 0;
                } catch (KafkaException e) {
                    // abort the transaction
                    Utils.printOut("Aborting transaction: %s", e.getMessage());
                    producer.abortTransaction();
                    retries = maybeRetry(retries, consumer);
                }

                remainingRecords = getRemainingRecords(consumer);
                if (remainingRecords != Long.MAX_VALUE) {
                    Utils.printOut("Remaining records: %d", remainingRecords);
                }
            }
        } catch (Throwable e) {
            Utils.printErr("Unhandled exception");
            e.printStackTrace();
        }
        Utils.printOut("Processed %d records", processedRecords);
        shutdown();
    }
This class implements a read-process-write application.
examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java