Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Normalize consumer group ID to find configuration bean #876

Merged
merged 5 commits into from
Sep 29, 2023
Merged

Normalize consumer group ID to find configuration bean #876

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
6667ca8
Add test
guillermocalvo Sep 20, 2023
2d479c6
Normalize consumer group ID to find configuration bean if need be
guillermocalvo Sep 20, 2023
239e8a6
Suppress warning
guillermocalvo Sep 20, 2023
1adc18e
Refactor for readability
guillermocalvo Sep 26, 2023
5e1cb34
Merge branch 'main' into 352-mechanism-of-obtaining-consumer-settings…
sdelamo Sep 29, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion kafka/src/main/java/io/micronaut/configuration/kafka/processor/KafkaConsumerProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -314,7 +314,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))
final AbstractKafkaConsumerConfiguration<?, ?> consumerConfigurationDefaults = getConsumerConfigurationBean(groupId)
.orElse(defaultConsumerConfiguration);
guillermocalvo marked this conversation as resolved.
Show resolved Hide resolved
if (consumerAnnotation.isTrue("uniqueGroupId")) {
groupId = groupId + "_" + UUID.randomUUID();
Expand Down Expand Up @@ -353,6 +353,13 @@ public void close() {
consumers.clear();
}

@SuppressWarnings("rawtypes")
private Optional<AbstractKafkaConsumerConfiguration> getConsumerConfigurationBean(String groupId) {
return beanContext.findBean(AbstractKafkaConsumerConfiguration.class, Qualifiers.byName(groupId))
.or(() -> NameUtils.isValidHyphenatedPropertyName(groupId) ? Optional.empty() :
beanContext.findBean(AbstractKafkaConsumerConfiguration.class, Qualifiers.byName(NameUtils.hyphenate(groupId))));
}

private Properties createConsumerProperties(final AnnotationValue<KafkaListener> consumerAnnotation,
final DefaultKafkaConsumerConfiguration consumerConfiguration,
final String clientId,
Expand Down
39 changes: 39 additions & 0 deletions kafka/src/test/groovy/io/micronaut/configuration/kafka/KafkaConfigurationSpec.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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 {
Expand Down