Spring/spring-boot
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add Kafka Streams auto-configuration

Browse Source 
See gh-14021
This commit is contained in:
Gary Russell
2018-08-08 14:42:15 -04:00
committed by Stephane Nicoll
parent f9207dd946
commit a7acbbd625
6 changed files with 349 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfiguration.java
View File

@@ -17,7 +17,11 @@
package org.springframework.boot.autoconfigure.kafka;
import java.io.IOException;
import java.util.Map;
import org.apache.kafka.streams.StreamsBuilder;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
@@ -28,12 +32,17 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.core.env.Environment;
import org.springframework.kafka.annotation.EnableKafkaStreams;
import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
import org.springframework.kafka.config.KafkaStreamsConfiguration;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.core.StreamsBuilderFactoryBean;
import org.springframework.kafka.security.jaas.KafkaJaasLoginModuleInitializer;
import org.springframework.kafka.support.LoggingProducerListener;
import org.springframework.kafka.support.ProducerListener;
@@ -138,4 +147,57 @@ public class KafkaAutoConfiguration {
return kafkaAdmin;
}
@Configuration
@ConditionalOnClass(StreamsBuilder.class)
public static class KafkaStreamsAutoConfiguration {
@Bean(KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public KafkaStreamsConfiguration defaultKafkaStreamsConfig(
KafkaProperties properties, Environment environment) {
Map<String, Object> streamsProperties = properties.buildStreamsProperties();
if (properties.getStreams().getApplicationId() == null) {
if (environment.getProperty("spring.application.id") != null) {
streamsProperties.put("application.id",
environment.getProperty("spring.application.name"));
}
}
return new KafkaStreamsConfiguration(streamsProperties);
}
@Bean
public KafkaStreamsFactoryBeanConfigurer kafkaStreamsFactoryBeanConfigurer(
StreamsBuilderFactoryBean factoryBean, KafkaProperties properties) {
return new KafkaStreamsFactoryBeanConfigurer(factoryBean, properties);
}
@Configuration
@EnableKafkaStreams
public static class EnableKafkaStreamsAutoConfiguration {
}
static class KafkaStreamsFactoryBeanConfigurer implements InitializingBean {
private final StreamsBuilderFactoryBean factoryBean;
private final KafkaProperties properties;
KafkaStreamsFactoryBeanConfigurer(StreamsBuilderFactoryBean factoryBean,
KafkaProperties properties) {
this.factoryBean = factoryBean;
this.properties = properties;
}
@Override
public void afterPropertiesSet() throws Exception {
this.factoryBean
.setAutoStartup(this.properties.getStreams().isAutoStartup());
}
}
}
}

156
spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java
View File

@@ -57,7 +57,7 @@ public class KafkaProperties {
/**
* Comma-delimited list of host:port pairs to use for establishing the initial
* connection to the Kafka cluster.
* connection to the Kafka cluster. Applies to all components unless overridden.
*/
private List<String> bootstrapServers = new ArrayList<>(
Collections.singletonList("localhost:9092"));
@@ -79,6 +79,8 @@ public class KafkaProperties {
private final Admin admin = new Admin();
private final Streams streams = new Streams();
private final Listener listener = new Listener();
private final Ssl ssl = new Ssl();
@@ -123,6 +125,10 @@ public class KafkaProperties {
return this.admin;
}
public Streams getStreams() {
return this.streams;
}
public Ssl getSsl() {
return this.ssl;
}
@@ -193,6 +199,19 @@ public class KafkaProperties {
return properties;
}
/**
* Create an initial map of streams properties from the state of this instance.
* <p>
* This allows you to add additional properties, if necessary.
* @return the streams properties initialized with the customizations defined on this
* instance
*/
public Map<String, Object> buildStreamsProperties() {
Map<String, Object> properties = buildCommonProperties();
properties.putAll(this.streams.buildProperties());
return properties;
}
public static class Consumer {
private final Ssl ssl = new Ssl();
@@ -211,7 +230,7 @@ public class KafkaProperties {
/**
* Comma-delimited list of host:port pairs to use for establishing the initial
* connection to the Kafka cluster.
* connection to the Kafka cluster. Overrides the global property, for consumers.
*/
private List<String> bootstrapServers;
@@ -421,7 +440,7 @@ public class KafkaProperties {
/**
* Comma-delimited list of host:port pairs to use for establishing the initial
* connection to the Kafka cluster.
* connection to the Kafka cluster. Overrides the global property, for producers.
*/
private List<String> bootstrapServers;
@@ -631,6 +650,136 @@ public class KafkaProperties {
}
/**
* High (and some medium) priority Streams properties and a general properties bucket.
*/
public static class Streams {
private final Ssl ssl = new Ssl();
/**
* Kafka streams application.id property; default spring.application.name.
*/
private String applicationId;
/**
* Whether or not to auto-start the streams factory bean.
*/
private boolean autoStartup;
/**
* Comma-delimited list of host:port pairs to use for establishing the initial
* connection to the Kafka cluster. Overrides the global property, for streams.
*/
private List<String> bootstrapServers;
/**
* Maximum number of memory bytes to be used for buffering across all threads.
*/
private Integer cacheMaxBytesBuffering;
/**
* ID to pass to the server when making requests. Used for server-side logging.
*/
private String clientId;
/**
* The replication factor for change log topics and repartition topics created by
* the stream processing application.
*/
private Integer replicationFactor;
/**
* Directory location for the state store.
*/
private String stateDir;
/**
* Additional Kafka properties used to configure the streams.
*/
private final Map<String, String> properties = new HashMap<>();
public Ssl getSsl() {
return this.ssl;
}
public String getApplicationId() {
return this.applicationId;
}
public void setApplicationId(String applicationId) {
this.applicationId = applicationId;
}
public boolean isAutoStartup() {
return this.autoStartup;
}
public void setAutoStartup(boolean autoStartup) {
this.autoStartup = autoStartup;
}
public List<String> getBootstrapServers() {
return this.bootstrapServers;
}
public void setBootstrapServers(List<String> bootstrapServers) {
this.bootstrapServers = bootstrapServers;
}
public Integer getCacheMaxBytesBuffering() {
return this.cacheMaxBytesBuffering;
}
public void setCacheMaxBytesBuffering(Integer cacheMaxBytesBuffering) {
this.cacheMaxBytesBuffering = cacheMaxBytesBuffering;
}
public String getClientId() {
return this.clientId;
}
public void setClientId(String clientId) {
this.clientId = clientId;
}
public Integer getReplicationFactor() {
return this.replicationFactor;
}
public void setReplicationFactor(Integer replicationFactor) {
this.replicationFactor = replicationFactor;
}
public String getStateDir() {
return this.stateDir;
}
public void setStateDir(String stateDir) {
this.stateDir = stateDir;
}
public Map<String, String> getProperties() {
return this.properties;
}
public Map<String, Object> buildProperties() {
Properties properties = new Properties();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(this::getApplicationId).to(properties.in("application.id"));
map.from(this::getBootstrapServers)
.to(properties.in(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG));
map.from(this::getCacheMaxBytesBuffering)
.to(properties.in("cache.max.bytes.buffering"));
map.from(this::getClientId)
.to(properties.in(CommonClientConfigs.CLIENT_ID_CONFIG));
map.from(this::getReplicationFactor).to(properties.in("replication.factor"));
map.from(this::getStateDir).to(properties.in("state.dir"));
return properties.with(this.ssl, this.properties);
}
}
public static class Template {
/**
@@ -1011,6 +1160,7 @@ public class KafkaProperties {
}
@SuppressWarnings("serial")
private static class Properties extends HashMap<String, Object> {
public <V> java.util.function.Consumer<V> in(String key) {

66
spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java
View File

@@ -19,6 +19,7 @@ package org.springframework.boot.autoconfigure.kafka;
import java.io.File;
import java.util.Collections;
import java.util.Map;
import java.util.Properties;
import javax.security.auth.login.AppConfigurationEntry;
@@ -30,6 +31,7 @@ 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.apache.kafka.streams.StreamsConfig;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
@@ -37,8 +39,10 @@ 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.annotation.KafkaStreamsDefaultConfiguration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaStreamsConfiguration;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaAdmin;
@@ -273,6 +277,68 @@ public class KafkaAutoConfigurationTests {
});
}
@Test
public void streamsProperties() {
this.contextRunner.withPropertyValues("spring.kafka.clientId=cid",
"spring.kafka.bootstrap-servers=localhost:9092,localhost:9093",
"spring.application.name=appName",
"spring.kafka.properties.foo.bar.baz=qux.fiz.buz",
"spring.kafka.streams.cache-max-bytes-buffering=42",
"spring.kafka.streams.client-id=override",
"spring.kafka.streams.properties.fiz.buz=fix.fox",
"spring.kafka.streams.replication-factor=2",
"spring.kafka.streams.state-dir=/tmp/state",
"spring.kafka.streams.ssl.key-password=p7",
"spring.kafka.streams.ssl.key-store-location=classpath:ksLocP",
"spring.kafka.streams.ssl.key-store-password=p8",
"spring.kafka.streams.ssl.key-store-type=PKCS12",
"spring.kafka.streams.ssl.trust-store-location=classpath:tsLocP",
"spring.kafka.streams.ssl.trust-store-password=p9",
"spring.kafka.streams.ssl.trust-store-type=PKCS12",
"spring.kafka.streams.ssl.protocol=TLSv1.2").run((context) -> {
Properties configs = context.getBean(
KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME,
KafkaStreamsConfiguration.class).asProperties();
assertThat(configs.get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG))
.isEqualTo("localhost:9092, localhost:9093");
assertThat(
configs.get(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG))
.isEqualTo("42");
assertThat(configs.get(StreamsConfig.CLIENT_ID_CONFIG))
.isEqualTo("override");
assertThat(configs.get(StreamsConfig.REPLICATION_FACTOR_CONFIG))
.isEqualTo("2");
assertThat(configs.get(StreamsConfig.STATE_DIR_CONFIG))
.isEqualTo("/tmp/state");
assertThat(configs.get(SslConfigs.SSL_KEY_PASSWORD_CONFIG))
.isEqualTo("p7");
assertThat(
(String) configs.get(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG))
.endsWith(File.separator + "ksLocP");
assertThat(configs.get(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG))
.isEqualTo("p8");
assertThat(configs.get(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG))
.isEqualTo("PKCS12");
assertThat((String) configs
.get(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG))
.endsWith(File.separator + "tsLocP");
assertThat(configs.get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG))
.isEqualTo("p9");
assertThat(configs.get(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG))
.isEqualTo("PKCS12");
assertThat(configs.get(SslConfigs.SSL_PROTOCOL_CONFIG))
.isEqualTo("TLSv1.2");
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(context.getBean(
KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_BUILDER_BEAN_NAME))
.isNotNull();
});
}
@SuppressWarnings("unchecked")
@Test
public void listenerProperties() {