Off By One Offset

A consumer applies each record and commits after every one. Partition 3 is at offsets 100 through 104 when the pod is killed by an OOM at the moment marked below.

while (true) {
  ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
  for (ConsumerRecord<String, String> r : records) {
    apply(r);
    consumer.commitSync(Map.of(
        new TopicPartition(r.topic(), r.partition()),
        new OffsetAndMetadata(r.offset())));
  }
}

step	offset	`apply` ran	committed value written
1	100	yes	100
2	101	yes	101
3	102	yes	102
4	103	yes	103
5	104	yes	104
6	-	-	pod killed here

A new member takes over partition 3. How many records does it apply again?

It depends on auto.commit.interval.ms, because manual commits are buffered and flushed on that schedule

One: offset 104, because the committed value is the offset the consumer will resume from, so committing r.offset() instead of r.offset() + 1 replays the record that was just handled

None, because offset 104 was committed after apply returned and the group resumes at the record after the committed offset

Five, offsets 100 through 104, because commitSync inside the record loop is discarded when the poll loop is interrupted before the batch completes