private int maybeRetry(int retries, KafkaConsumer<Integer, String> consumer) {
        if (retries < 0) {
            Utils.printErr("The number of retries must be greater than zero");
            shutdown();
        }
        
        if (retries < MAX_RETRIES) {
            // retry: reset fetch offset
            // the consumer fetch position needs to be restored to the committed offset before the transaction started
            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(Collections.singleton(tp));
                }
            });
            retries++;
        } else {
            // continue: skip records
            // the consumer fetch position needs to be committed as if records were processed successfully
            Utils.printErr("Skipping records after %d retries", MAX_RETRIES);
            consumer.commitSync();
            retries = 0;
        }
        return retries;
    }
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