A manual offset commit lets a Kafka consumer persist offsets at an application-chosen point. The consumer processes records first, obtains the offsets representing completed work, and then commits them. section 4.3.5
Use the commit boundary deliberately:
- Commit after successful processing to favor at-least-once delivery; a restart may reprocess records after the latest committed offset. section 4.3.5
- Commit immediately after retrieving records to favor at-most-once delivery, accepting possible record loss if processing fails afterward. section 4.3.5
- Avoid committing offsets for asynchronous work that has not finished; the poll loop can otherwise advance beyond successfully processed records. section 4.3.5
Commit after processing
ConsumerRecords<String, ProductTransaction> records = consumer.poll(Duration.ofSeconds(5)); if (!records.isEmpty()) { recordProcessor.processRecords(records); Map<TopicPartition, OffsetAndMetadata> offsets = recordProcessor.getOffsets(); if (offsets != null) consumer.commitSync(offsets); } This sequence ties persistence to the offsets returned by the processor. section 4.3.5
Listing 4.4 Consumer code
ConsumerRecords<String, ProductTransaction> consumerRecords = consumer.poll(
Duration.ofSeconds(5));
if (!consumerRecords.isEmpty()) {
recordProcessor.processRecords(consumerRecords); #1
Map<TopicPartition, OffsetAndMetadata> offsetsAndMetadata =
recordProcessor.getOffsets(); #2
if (offsetsAndMetadata != null) {
consumer.commitSync(offsetsAndMetadata); #3
}
Use the consumer listing to inspect the poll, processing, offset collection, and synchronous commit sequence. section 4.3.5
