Page History

KafkaConsumer consumer = new KafkaConsumer<>(consumerConfig);
producer.initTransactions();
volatile boolean isRunning = true;

try {
    while (isRunning) {
        ConsumerRecords<String, String> records = consumer.poll(CONSUMER_POLL_TIMEOUT);
	        final boolean shouldCommit;
	        try {
           	 producer.beginTransaction();

		            // Do some processing and build the records we want to produce.
            		List<ProducerRecord> processed = process(consumed);

		            for (ProducerRecord record : processed)
                		producer.send(record, (metadata, exception) -> {
				                    // not required to capture the exception here.
			                });		
		
            producer.sendOffsetsToTransaction(consumedOffsets, consumer.groupMetadata());

		            shouldCommit = true;
 	        } catch (Exception e) {
		            // Catch any exception thrown from the data transmission phase.
		            shouldCommit = false;
 	        }

	        try {
            if (shouldCommit) {
                producer.commitTransaction();
            } else {
                resetToLastCommittedPositions(consumer);
                producer.abortTransaction();
            }
        } catch (CommitFailedException e) {
            // Transaction commit failed with abortable error, user could reset
            // the application state and resume with a new transaction. The root
            // cause was wrapped in the thrown exception.
   
         resetToLastCommittedPositions(consumer);
            producer.abortTransaction();
        }
    }
} catch (KafkaException e) {
     // Recommended closing producer/consumer for fatal exceptions
    producer.close();
        consumer.close();
        throw e;                   
    }
}