Unclear Micronaut Kafka documentation about commit strategy for SYNC_PER_RECORD

[jamesforwardnwboxed]

Issue description

Looking at: https://micronaut-projects.github.io/micronaut-kafka/latest/guide/#kafkaOffsets

We can see that SYNC says Commits offsets manually at the end of each poll() loop if no exceptions occur. Sets enable.auto.commit to false which explicitly implies that if an exception occurs offsets are not committed

However SYNC_PER_RECORD just says Commits offsets manually after each ConsumerRecord is processed. Sets enable.auto.commit to false without any clarification about exception handling.

2 questions:

1. Can the docs be updated to be clearer?
2. What happens in SYNC_PER_RECORD If an exception is thrown? And how does it relate to different ErrorStrategy values?

[graemerocher]

looks like auto commit is only enabled if offset strategy is AUTO

micronaut-kafka/kafka/src/main/java/io/micronaut/configuration/kafka/processor/KafkaConsumerProcessor.java

Line 414 in 9d442d2

properties.putIfAbsent(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, String.valueOf(offsetStrategy == OffsetStrategy.AUTO));

And it appears the same error strategy handling applies

micronaut-kafka/kafka/src/main/java/io/micronaut/configuration/kafka/processor/ConsumerStateSingle.java

Lines 90 to 104 in 9d442d2

	try {
	process(consumerRecord, consumerRecords);
	} catch (Exception e) {
	if (resolveWithErrorStrategy(consumerRecords, consumerRecord, e)) {
	resetTheFollowingPartitions(consumerRecord, iterator);
	failed = true;
	return;
	}
	}
	if (info.offsetStrategy == OffsetStrategy.SYNC_PER_RECORD) {
	commitSync(consumerRecords, consumerRecord, currentOffsets);
	} else if (info.offsetStrategy == OffsetStrategy.ASYNC_PER_RECORD) {
	kafkaConsumer.commitAsync(currentOffsets, this::resolveCommitCallback);
	}

Feel free to contribute all documentation clarifications that you feel are missing