Fix consumer interception state test regressions

Restore the multi-method ConsumerInfo constructor used by the new routing specs and stub pass-through interception in the state tests so they exercise the real dispatch path.

Co-Authored-By: Codex with GPT-5 <codex@openai.com>

Author: graemerocher

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

@@ -121,6 +121,17 @@ final class ConsumerInfo {
         );
     }
 
+    ConsumerInfo(
+        String clientId,
+        String groupId,
+        OffsetStrategy offsetStrategy,
+        AnnotationValue<KafkaListener> kafkaListener,
+        Properties properties,
+        List<ExecutableMethod<?, ?>> methods
+    ) {
+        this(clientId, groupId, offsetStrategy, kafkaListener, properties, methods, Map.of());
+    }
+
     @SuppressWarnings("unchecked")
     ConsumerInfo(
         String clientId,

kafka/src/test/groovy/io/micronaut/configuration/kafka/processor/ConsumerCreationStrategyStateSpec.groovy

@@ -30,6 +30,7 @@ class ConsumerCreationStrategyStateSpec extends Specification {
         def listener = new TestPerClassMultiTopicListener()
         def processor = Stub(KafkaConsumerProcessor) {
             getBinderRegistry() >> binderRegistry
+            interceptRecord(_, _ as ConsumerRecord) >> { ConsumerInfo ignored, ConsumerRecord<?, ?> record -> record }
         }
         def kafkaConsumer = Stub(Consumer) {
             subscription() >> ([] as Set)
@@ -54,6 +55,7 @@ class ConsumerCreationStrategyStateSpec extends Specification {
         def listener = new TestPerClassMultiTopicBatchListener()
         def processor = Stub(KafkaConsumerProcessor) {
             getBatchBinderRegistry() >> batchBinderRegistry
+            interceptRecords(_, _ as ConsumerRecords) >> { ConsumerInfo ignored, ConsumerRecords<?, ?> records -> records }
         }
         def kafkaConsumer = Stub(Consumer) {
             subscription() >> ([] as Set)
@@ -78,6 +80,7 @@ class ConsumerCreationStrategyStateSpec extends Specification {
         def listener = new TestBatchMultiTopicListener()
         def processor = Stub(KafkaConsumerProcessor) {
             getBatchBinderRegistry() >> batchBinderRegistry
+            interceptRecords(_, _ as ConsumerRecords) >> { ConsumerInfo ignored, ConsumerRecords<?, ?> records -> records }
         }
         def kafkaConsumer = Stub(Consumer) {
             subscription() >> ([] as Set)