When is offset committed in KafkaListener.java in Springboot?

Question

There are two parts to my question.

1. In below Springboot KafkaListener implementation when does offset get committed for offset strategy auto-offset-reset: latest and enable-auto-commit: true? Is it immediately after the message is received by the consumer or after the whole method implementing the KafkaListener has completed?

KafkaConsumer.java

@KafkaListener(topics = "${spring.kafka.consumer.topic}")
    public ResponseEntity<String> consume(String message) {
        log.info("Message recieved from Kafka topic {}", message); // offset committed HERE?
        KafkaResponse kafkaResponse = new Gson().fromJson(message, KafkaResponse.class);
        myBusinessService.processKafkaResponse(kafkaResponse);
        return new ResponseEntity<>("Successfully Received", HttpStatus.OK);// OR offset committed HERE?
    }

2. In application.yml for the below properties which statements are True/False/what's the correct answer?:

max-poll-records: 100
max-poll-interval-ms: 200000
enable-auto-commit: true 
auto-commit-interval: 3000
auto-offset-reset: latest
isolation-level: READ_UNCOMMITTED
fetch-max-bytes: 52428800

• Between auto-offset-reset: latest and auto-commit-interval: 3000 if the Kafka consumer breaks down before 3 seconds then no offset would be committed for any record processed during those 3 seconds?
• Between max-poll-interval-ms: 200000 and auto-commit-interval: 3000 Kafka consumer would surely be polling the broker after an interval of 200000 seconds even if the consumer hasn't finished committing the current batch of offsets in 3000 seconds?
• Between the combination of max-poll-records: 100 and fetch-max-bytes: 52428800 which would take a priority if 99th record has exceeded 52428800 bytes?

Thanks in advance!

Answer 1

It's best not to use enable.auto.commit=true. Spring commits the offsets in a more deterministic fashion, either after all the records from a poll have been processed (default - AckMode.BATCH) or after each record is processed AckMode.RECORD.

enable.auto.commit won't commit until the next poll() and then only if the the auto.commit.interval has passed.