Apply review feedback: fix TopicPartition bucketing, empty batch commit path, error record selection, and docs typo

Agent-Logs-Url: https://github.com/micronaut-projects/micronaut-kafka/sessions/d513a696-a1bd-4396-a074-79da5f5ae36b

Co-authored-by: graemerocher <66626+graemerocher@users.noreply.github.com>

Author: Copilot

kafka/src/main/java/io/micronaut/configuration/kafka/processor/ConsumerStateBatch.java

@@ -92,6 +92,7 @@ protected void processRecords(ConsumerRecords<?, ?> consumerRecords, @Nullable M
             }
             final ConsumerRecords<?, ?> interceptedConsumerRecords = kafkaConsumerProcessor.interceptRecords(info, consumerRecords);
             if (interceptedConsumerRecords.isEmpty()) {
+                handleResult(normalizeResult(null), consumerRecords);
                 failed = false;
                 return;
             }

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

@@ -88,14 +88,16 @@ protected void processRecords(ConsumerRecords<?, ?> consumerRecords,
             Optional.ofNullable(info.seekArg).ifPresent(argument -> boundArguments.put(argument, seek));
             Optional.ofNullable(info.ackArg).ifPresent(argument -> boundArguments.put(argument, (KafkaAcknowledgement) () -> kafkaConsumer.commitSync(currentOffsets)));
 
+            ConsumerRecord<?, ?> interceptedConsumerRecord = null;
             try {
-                ConsumerRecord<?, ?> interceptedConsumerRecord = kafkaConsumerProcessor.interceptRecord(info, consumerRecord);
+                interceptedConsumerRecord = kafkaConsumerProcessor.interceptRecord(info, consumerRecord);
                 if (interceptedConsumerRecord != null) {
                     process(interceptedConsumerRecord, consumerRecords);
                 }
             } catch (Exception e) {
-                if (resolveWithErrorStrategy(consumerRecords, consumerRecord, e)) {
-                    resetTheFollowingPartitions(consumerRecord, iterator);
+                final ConsumerRecord<?, ?> errorRecord = interceptedConsumerRecord != null ? interceptedConsumerRecord : consumerRecord;
+                if (resolveWithErrorStrategy(consumerRecords, errorRecord, e)) {
+                    resetTheFollowingPartitions(errorRecord, iterator);
                     failed = true;
                     return;
                 }

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

@@ -455,7 +455,7 @@ <K, V> ConsumerRecords<K, V> interceptRecords(@NonNull ConsumerInfo consumerInfo
         for (ConsumerRecord<K, V> consumerRecord : consumerRecords) {
             ConsumerRecord<K, V> intercepted = interceptRecord(consumerInfo, consumerRecord);
             if (intercepted != null) {
-                TopicPartition topicPartition = new TopicPartition(consumerRecord.topic(), consumerRecord.partition());
+                TopicPartition topicPartition = new TopicPartition(intercepted.topic(), intercepted.partition());
                 interceptedRecords.computeIfAbsent(topicPartition, ignored -> new ArrayList<>()).add(intercepted);
             }
         }

src/main/docs/guide/kafkaListener/kafkaListenerInterceptors.adoc

@@ -1,4 +1,4 @@
-To inspect, filter, or wrap consumed records before Micronaut binds them to a ann:configuration.kafka.annotation.KafkaListener[] method, define a bean that implements api:configuration.kafka.ConsumerRecordInterceptor[].
+To inspect, filter, or wrap consumed records before Micronaut binds them to an ann:configuration.kafka.annotation.KafkaListener[] method, define a bean that implements api:configuration.kafka.ConsumerRecordInterceptor[].
 
 Interceptors run before argument binding for single-record listeners and before Micronaut builds the batch arguments for batch listeners. If multiple interceptor beans are present, Micronaut applies them in order.