Normalize consumer group ID to find configuration bean

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

@@ -291,8 +291,7 @@ public void process(BeanDefinition<?> beanDefinition, ExecutableMethod<?, ?> met
                 .orElseGet(() -> applicationConfiguration.getName().map(s -> s + '-' + NameUtils.hyphenate(beanType.getSimpleName())).orElse(null));
         final OffsetStrategy offsetStrategy = consumerAnnotation.enumValue("offsetStrategy", OffsetStrategy.class)
                 .orElse(OffsetStrategy.AUTO);
-        final AbstractKafkaConsumerConfiguration<?, ?> consumerConfigurationDefaults = beanContext.findBean(AbstractKafkaConsumerConfiguration.class, Qualifiers.byName(groupId))
-                .orElse(defaultConsumerConfiguration);
+        final AbstractKafkaConsumerConfiguration<?, ?> consumerConfigurationDefaults = getConsumerConfigurationDefaults(groupId);
         if (consumerAnnotation.isTrue("uniqueGroupId")) {
             groupId = groupId + "_" + UUID.randomUUID();
         }
@@ -361,6 +360,26 @@ <T, R> BoundExecutable<T, R> bindAsBatch(ExecutableMethod<T, R> method, Map<Argu
         return batchBinder.bind(method, batchBinderRegistry, consumerRecords);
     }
 
+    @SuppressWarnings("rawtypes")
+    private AbstractKafkaConsumerConfiguration getConsumerConfigurationDefaults(String groupId) {
+        return findConfigurationBean(groupId)
+            .or(() -> findHyphenatedConsumerConfigurationBean(groupId))
+            .orElse(defaultConsumerConfiguration);
+    }
+
+    @SuppressWarnings("rawtypes")
+    private Optional<AbstractKafkaConsumerConfiguration> findConfigurationBean(String groupId) {
+        return beanContext.findBean(AbstractKafkaConsumerConfiguration.class, Qualifiers.byName(groupId));
+    }
+
+    @SuppressWarnings("rawtypes")
+    private Optional<AbstractKafkaConsumerConfiguration> findHyphenatedConsumerConfigurationBean(String groupId) {
+        if (NameUtils.isValidHyphenatedPropertyName(groupId)) {
+            return Optional.empty();
+        }
+        return findConfigurationBean(NameUtils.hyphenate(groupId));
+    }
+
     @SuppressWarnings("rawtypes")
     private Properties createConsumerProperties(final AnnotationValue<KafkaListener> consumerAnnotation,
                                                 final DefaultKafkaConsumerConfiguration consumerConfiguration,

kafka/src/test/groovy/io/micronaut/configuration/kafka/KafkaConfigurationSpec.groovy

@@ -1,10 +1,13 @@
 package io.micronaut.configuration.kafka
 
+import io.micronaut.configuration.kafka.annotation.KafkaListener
+import io.micronaut.configuration.kafka.annotation.Topic
 import io.micronaut.configuration.kafka.config.AbstractKafkaConfiguration
 import io.micronaut.configuration.kafka.config.AbstractKafkaConsumerConfiguration
 import io.micronaut.configuration.kafka.config.AbstractKafkaProducerConfiguration
 import io.micronaut.configuration.kafka.config.KafkaConsumerConfiguration
 import io.micronaut.context.ApplicationContext
+import io.micronaut.context.annotation.Requires
 import io.micronaut.context.env.EnvironmentPropertySource
 import io.micronaut.context.env.MapPropertySource
 import io.micronaut.context.exceptions.NoSuchBeanException
@@ -14,6 +17,7 @@ import org.apache.kafka.clients.consumer.KafkaConsumer
 import org.apache.kafka.clients.producer.KafkaProducer
 import org.apache.kafka.clients.producer.Producer
 import org.apache.kafka.clients.producer.ProducerConfig
+import org.apache.kafka.common.serialization.IntegerDeserializer
 import org.apache.kafka.common.serialization.StringDeserializer
 import org.apache.kafka.common.serialization.StringSerializer
 import spock.lang.AutoCleanup
@@ -164,6 +168,35 @@ class KafkaConfigurationSpec extends Specification {
         consumer.close()
     }
 
+    void "test consumer with camel-case group id"() {
+        given:
+        applicationContext = ApplicationContext.run(
+                'spec.name': 'KafkaConfigurationSpec',
+                ('kafka.' + ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG): "localhost:1111",
+                ('kafka.consumers.my-kebab-group.' + ConsumerConfig.GROUP_ID_CONFIG): "my-kebab-group",
+                ('kafka.consumers.my-kebab-group.' + ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG): IntegerDeserializer.name,
+                ('kafka.consumers.my-kebab-group.' + ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG): StringDeserializer.name
+        )
+
+        when:
+        MyConsumer consumer = applicationContext.getBean(MyConsumer)
+
+        then:
+        consumer != null
+
+        when:
+        KafkaConsumer kafkaConsumer = consumer.kafkaConsumer
+
+        then:
+        kafkaConsumer != null
+        kafkaConsumer.groupId.orElse(null) == 'MY_KEBAB_GROUP'
+        kafkaConsumer.keyDeserializer instanceof IntegerDeserializer
+        kafkaConsumer.valueDeserializer instanceof StringDeserializer
+
+        cleanup:
+        applicationContext.close()
+    }
+
     void "test configure list fields default properties"() {
         given:
         applicationContext = ApplicationContext.run(
@@ -319,6 +352,12 @@ class KafkaConfigurationSpec extends Specification {
         thrown(NoSuchBeanException)
     }
 
+    @Requires(property = 'spec.name', value = 'KafkaConfigurationSpec')
+    @KafkaListener(groupId = "MY_KEBAB_GROUP", autoStartup = false)
+    static class MyConsumer implements ConsumerAware<String, String> {
+        Consumer<String, String> kafkaConsumer
+        @Topic("foo") void consume(String foo) { }
+    }
 }
 
 class FakeYamlPropertySource extends MapPropertySource {