root
/
spring-boot
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

Auto-configure Kafka MessageConverter

Browse Source 
See gh-10380
pull/10494/head
Eddú Meléndez 7 years ago committed by Stephane Nicoll
parent 41424e4529
commit d7bc93f278
4 changed files with 271 additions and 179 deletions
Show Stats Download Patch File Download Diff File

15
spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurer.java
Unescape Escape View File

@ -20,17 +20,21 @@ import org.springframework.boot.autoconfigure.kafka.KafkaProperties.Listener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.support.converter.MessageConverter;
/**
* Configure {@link ConcurrentKafkaListenerContainerFactory} with sensible defaults.
*
* @author Gary Russell
* @author Eddú Meléndez
* @since 1.5.0
*/
public class ConcurrentKafkaListenerContainerFactoryConfigurer {
private KafkaProperties properties;
private MessageConverter messageConverter;
/**
* Set the {@link KafkaProperties} to use.
* @param properties the properties
@ -39,6 +43,14 @@ public class ConcurrentKafkaListenerContainerFactoryConfigurer {
this.properties = properties;
}
/**
* Set the {@link MessageConverter} to use.
* @param messageConverter the message converter
*/
public void setMessageConverter(MessageConverter messageConverter) {
this.messageConverter = messageConverter;
}
/**
* Configure the specified Kafka listener container factory. The factory can be
* further tuned and default settings can be overridden.
@ -49,6 +61,9 @@ public class ConcurrentKafkaListenerContainerFactoryConfigurer {
public void configure(
ConcurrentKafkaListenerContainerFactory<Object, Object> listenerContainerFactory,
ConsumerFactory<Object, Object> consumerFactory) {
if (this.messageConverter != null) {
listenerContainerFactory.setMessageConverter(this.messageConverter);
}
listenerContainerFactory.setConsumerFactory(consumerFactory);
Listener container = this.properties.getListener();
ContainerProperties containerProperties = listenerContainerFactory

10
spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAnnotationDrivenConfiguration.java
Unescape Escape View File

@ -16,6 +16,7 @@
package org.springframework.boot.autoconfigure.kafka;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
@ -24,11 +25,13 @@ import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerConfigUtils;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.support.converter.MessageConverter;
/**
* Configuration for Kafka annotation-driven support.
*
* @author Gary Russell
* @author Eddú Meléndez
* @since 1.5.0
*/
@Configuration
@ -37,8 +40,12 @@ class KafkaAnnotationDrivenConfiguration {
private final KafkaProperties properties;
KafkaAnnotationDrivenConfiguration(KafkaProperties properties) {
private final MessageConverter messageConverter;
KafkaAnnotationDrivenConfiguration(KafkaProperties properties,
ObjectProvider<MessageConverter> messageConverter) {
this.properties = properties;
this.messageConverter = messageConverter.getIfAvailable();
}
@Bean
@ -46,6 +53,7 @@ class KafkaAnnotationDrivenConfiguration {
public ConcurrentKafkaListenerContainerFactoryConfigurer kafkaListenerContainerFactoryConfigurer() {
ConcurrentKafkaListenerContainerFactoryConfigurer configurer = new ConcurrentKafkaListenerContainerFactoryConfigurer();
configurer.setKafkaProperties(this.properties);
configurer.setMessageConverter(this.messageConverter);
return configurer;
}

12
spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfiguration.java
Unescape Escape View File

@ -18,6 +18,7 @@ package org.springframework.boot.autoconfigure.kafka;
import java.io.IOException;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
@ -36,12 +37,14 @@ import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.security.jaas.KafkaJaasLoginModuleInitializer;
import org.springframework.kafka.support.LoggingProducerListener;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.kafka.support.converter.RecordMessageConverter;
/**
* {@link EnableAutoConfiguration Auto-configuration} for Apache Kafka.
*
* @author Gary Russell
* @author Stephane Nicoll
* @author Eddú Meléndez
* @since 1.5.0
*/
@Configuration
@ -52,8 +55,12 @@ public class KafkaAutoConfiguration {
private final KafkaProperties properties;
public KafkaAutoConfiguration(KafkaProperties properties) {
private final RecordMessageConverter messageConverter;
public KafkaAutoConfiguration(KafkaProperties properties,
ObjectProvider<RecordMessageConverter> messageConverter) {
this.properties = properties;
this.messageConverter = messageConverter.getIfAvailable();
}
@Bean
@ -63,6 +70,9 @@ public class KafkaAutoConfiguration {
ProducerListener<Object, Object> kafkaProducerListener) {
KafkaTemplate<Object, Object> kafkaTemplate = new KafkaTemplate<>(
kafkaProducerFactory);
if (this.messageConverter != null) {
kafkaTemplate.setMessageConverter(this.messageConverter);
}
kafkaTemplate.setProducerListener(kafkaProducerListener);
kafkaTemplate.setDefaultTopic(this.properties.getTemplate().getDefaultTopic());
return kafkaTemplate;

137
spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java
Unescape Escape View File

@ -30,12 +30,14 @@ import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.LongSerializer;
import org.junit.After;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.boot.test.util.TestPropertyValues;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
@ -43,31 +45,33 @@ import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode;
import org.springframework.kafka.security.jaas.KafkaJaasLoginModuleInitializer;
import org.springframework.kafka.support.converter.MessageConverter;
import org.springframework.kafka.support.converter.MessagingMessageConverter;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.entry;
import static org.mockito.Mockito.mock;
/**
* Tests for {@link KafkaAutoConfiguration}.
*
* @author Gary Russell
* @author Stephane Nicoll
* @author Eddú Meléndez
*/
public class KafkaAutoConfigurationTests {
private AnnotationConfigApplicationContext context;
@After
public void closeContext() {
if (this.context != null) {
this.context.close();
}
}
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(KafkaAutoConfiguration.class));
@Test
public void consumerProperties() {
load("spring.kafka.bootstrap-servers=foo:1234", "spring.kafka.properties.foo=bar",
this.contextRunner.withUserConfiguration(TestConfiguration.class)
.withPropertyValues(
"spring.kafka.bootstrap-servers=foo:1234",
"spring.kafka.properties.foo=bar",
"spring.kafka.properties.baz=qux",
"spring.kafka.properties.foo.bar.baz=qux.fiz.buz",
"spring.kafka.ssl.key-password=p1",
@ -86,8 +90,9 @@ public class KafkaAutoConfigurationTests {
"spring.kafka.consumer.group-id=bar",
"spring.kafka.consumer.heartbeat-interval=234",
"spring.kafka.consumer.key-deserializer = org.apache.kafka.common.serialization.LongDeserializer",
"spring.kafka.consumer.value-deserializer = org.apache.kafka.common.serialization.IntegerDeserializer");
DefaultKafkaConsumerFactory<?, ?> consumerFactory = this.context
"spring.kafka.consumer.value-deserializer = org.apache.kafka.common.serialization.IntegerDeserializer")
.run((context) -> {
DefaultKafkaConsumerFactory<?, ?> consumerFactory = context
.getBean(DefaultKafkaConsumerFactory.class);
Map<String, Object> configs = consumerFactory.getConfigurationProperties();
// common
@ -123,14 +128,19 @@ public class KafkaAutoConfigurationTests {
assertThat(configs.get("baz")).isEqualTo("qux");
assertThat(configs.get("foo.bar.baz")).isEqualTo("qux.fiz.buz");
assertThat(configs.get("fiz.buz")).isEqualTo("fix.fox");
});
}
@Test
public void producerProperties() {
load("spring.kafka.clientId=cid",
this.contextRunner.withUserConfiguration(TestConfiguration.class)
.withPropertyValues(
"spring.kafka.clientId=cid",
"spring.kafka.properties.foo.bar.baz=qux.fiz.buz",
"spring.kafka.producer.acks=all", "spring.kafka.producer.batch-size=20",
"spring.kafka.producer.bootstrap-servers=bar:1234", // test override
"spring.kafka.producer.acks=all",
"spring.kafka.producer.batch-size=20",
"spring.kafka.producer.bootstrap-servers=bar:1234", // test
// override
"spring.kafka.producer.buffer-memory=12345",
"spring.kafka.producer.compression-type=gzip",
"spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.LongSerializer",
@ -141,8 +151,9 @@ public class KafkaAutoConfigurationTests {
"spring.kafka.producer.ssl.keystore-password=p5",
"spring.kafka.producer.ssl.truststore-location=classpath:tsLocP",
"spring.kafka.producer.ssl.truststore-password=p6",
"spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.IntegerSerializer");
DefaultKafkaProducerFactory<?, ?> producerFactory = this.context
"spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.IntegerSerializer")
.run((context) -> {
DefaultKafkaProducerFactory<?, ?> producerFactory = context
.getBean(DefaultKafkaProducerFactory.class);
Map<String, Object> configs = producerFactory.getConfigurationProperties();
// common
@ -167,15 +178,16 @@ public class KafkaAutoConfigurationTests {
assertThat(configs.get(ProducerConfig.RETRIES_CONFIG)).isEqualTo(2);
assertThat(configs.get(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG))
.isEqualTo(IntegerSerializer.class);
assertThat(this.context.getBeansOfType(KafkaJaasLoginModuleInitializer.class))
assertThat(context.getBeansOfType(KafkaJaasLoginModuleInitializer.class))
.isEmpty();
assertThat(configs.get("foo.bar.baz")).isEqualTo("qux.fiz.buz");
assertThat(configs.get("fiz.buz")).isEqualTo("fix.fox");
});
}
@Test
public void adminProperties() {
load("spring.kafka.clientId=cid",
this.contextRunner.withPropertyValues("spring.kafka.clientId=cid",
"spring.kafka.properties.foo.bar.baz=qux.fiz.buz",
"spring.kafka.admin.fail-fast=true",
"spring.kafka.admin.properties.fiz.buz=fix.fox",
@ -183,8 +195,8 @@ public class KafkaAutoConfigurationTests {
"spring.kafka.admin.ssl.keystore-location=classpath:ksLocP",
"spring.kafka.admin.ssl.keystore-password=p5",
"spring.kafka.admin.ssl.truststore-location=classpath:tsLocP",
"spring.kafka.admin.ssl.truststore-password=p6");
KafkaAdmin admin = this.context.getBean(KafkaAdmin.class);
"spring.kafka.admin.ssl.truststore-password=p6").run((context) -> {
KafkaAdmin admin = context.getBean(KafkaAdmin.class);
Map<String, Object> configs = admin.getConfig();
// common
assertThat(configs.get(AdminClientConfig.CLIENT_ID_CONFIG)).isEqualTo("cid");
@ -197,34 +209,40 @@ public class KafkaAutoConfigurationTests {
.endsWith(File.separator + "tsLocP");
assertThat(configs.get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG))
.isEqualTo("p6");
assertThat(this.context.getBeansOfType(KafkaJaasLoginModuleInitializer.class))
assertThat(context.getBeansOfType(KafkaJaasLoginModuleInitializer.class))
.isEmpty();
assertThat(configs.get("foo.bar.baz")).isEqualTo("qux.fiz.buz");
assertThat(configs.get("fiz.buz")).isEqualTo("fix.fox");
assertThat(KafkaTestUtils.getPropertyValue(admin, "fatalIfBrokerNotAvailable",
Boolean.class)).isTrue();
});
}
@SuppressWarnings("unchecked")
@Test
public void listenerProperties() {
load("spring.kafka.template.default-topic=testTopic",
this.contextRunner.withUserConfiguration(TestConfiguration.class)
.withPropertyValues("spring.kafka.template.default-topic=testTopic",
"spring.kafka.listener.ack-mode=MANUAL",
"spring.kafka.listener.ack-count=123",
"spring.kafka.listener.ack-time=456",
"spring.kafka.listener.concurrency=3",
"spring.kafka.listener.poll-timeout=2000",
"spring.kafka.listener.type=batch", "spring.kafka.jaas.enabled=true",
"spring.kafka.listener.type=batch",
"spring.kafka.jaas.enabled=true",
"spring.kafka.jaas.login-module=foo",
"spring.kafka.jaas.control-flag=REQUISITE",
"spring.kafka.jaas.options.useKeyTab=true");
DefaultKafkaProducerFactory<?, ?> producerFactory = this.context
"spring.kafka.jaas.options.useKeyTab=true")
.run((context) -> {
DefaultKafkaProducerFactory<?, ?> producerFactory = context
.getBean(DefaultKafkaProducerFactory.class);
DefaultKafkaConsumerFactory<?, ?> consumerFactory = this.context
DefaultKafkaConsumerFactory<?, ?> consumerFactory = context
.getBean(DefaultKafkaConsumerFactory.class);
KafkaTemplate<?, ?> kafkaTemplate = this.context.getBean(KafkaTemplate.class);
KafkaListenerContainerFactory<?> kafkaListenerContainerFactory = this.context
KafkaTemplate<?, ?> kafkaTemplate = context.getBean(KafkaTemplate.class);
KafkaListenerContainerFactory<?> kafkaListenerContainerFactory = context
.getBean(KafkaListenerContainerFactory.class);
assertThat(kafkaTemplate.getMessageConverter()).isInstanceOf(
MessagingMessageConverter.class);
assertThat(new DirectFieldAccessor(kafkaTemplate)
.getPropertyValue("producerFactory")).isEqualTo(producerFactory);
assertThat(kafkaTemplate.getDefaultTopic()).isEqualTo("testTopic");
@ -238,9 +256,9 @@ public class KafkaAutoConfigurationTests {
assertThat(dfa.getPropertyValue("containerProperties.pollTimeout"))
.isEqualTo(2000L);
assertThat(dfa.getPropertyValue("batchListener")).isEqualTo(true);
assertThat(this.context.getBeansOfType(KafkaJaasLoginModuleInitializer.class))
assertThat(context.getBeansOfType(KafkaJaasLoginModuleInitializer.class))
.hasSize(1);
KafkaJaasLoginModuleInitializer jaas = this.context
KafkaJaasLoginModuleInitializer jaas = context
.getBean(KafkaJaasLoginModuleInitializer.class);
dfa = new DirectFieldAccessor(jaas);
assertThat(dfa.getPropertyValue("loginModule")).isEqualTo("foo");
@ -248,14 +266,55 @@ public class KafkaAutoConfigurationTests {
.isEqualTo(AppConfigurationEntry.LoginModuleControlFlag.REQUISITE);
assertThat(((Map<String, String>) dfa.getPropertyValue("options")))
.containsExactly(entry("useKeyTab", "true"));
});
}
@Test
public void testKafkaTemplateRecordMessageConverters() {
this.contextRunner.withUserConfiguration(RecordMessageConvertersConfiguration.class)
.run((context) -> {
KafkaTemplate kafkaTemplate = context.getBean(KafkaTemplate.class);
assertThat(kafkaTemplate.getMessageConverter())
.isSameAs(context.getBean("myRecordMessageConverter"));
});
}
@Test
public void testConcurrentKafkaListenerContainerFactoryWithCustomMessageConverters() {
this.contextRunner.withUserConfiguration(MessageConvertersConfiguration.class)
.run((context) -> {
ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory = context
.getBean(ConcurrentKafkaListenerContainerFactory.class);
DirectFieldAccessor dfa = new DirectFieldAccessor(
kafkaListenerContainerFactory);
assertThat(dfa.getPropertyValue("messageConverter"))
.isSameAs(context.getBean("myMessageConverter"));
});
}
@Configuration
protected static class TestConfiguration {
}
@Configuration
protected static class RecordMessageConvertersConfiguration {
@Bean
public RecordMessageConverter myRecordMessageConverter() {
return mock(RecordMessageConverter.class);
}
}
@Configuration
protected static class MessageConvertersConfiguration {
@Bean
public MessageConverter myMessageConverter() {
return mock(MessageConverter.class);
}
private void load(String... environment) {
AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext();
ctx.register(KafkaAutoConfiguration.class);
TestPropertyValues.of(environment).applyTo(ctx);
ctx.refresh();
this.context = ctx;
}
}

Write Preview
Loading…
Cancel
Save