@Override
    public void close() throws Exception {
        if (producer != null) {
            producer.close();
        }

        if (consumer != null) {
            consumer.close();
        }
    }
When we get a generic {@code KafkaException} while processing records, we retry up to {@code MAX_RETRIES} times.
If we exceed this threshold, we log an error and move on to the next batch of records.
In a real world application you may want to to send these records to a dead letter topic (DLT) for further processing.

@param retries Current number of retries
@param consumer Consumer instance
@return Updated number of retries
examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java