Polishing - PR Comments

Author: garyrussell

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

@@ -17,11 +17,11 @@
 package org.springframework.boot.autoconfigure.kafka;
 
 import java.io.IOException;
-import java.util.List;
-import java.util.Properties;
+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;
@@ -32,8 +32,10 @@
 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;
@@ -147,33 +149,52 @@ public KafkaAdmin kafkaAdmin() {
 
 	@Configuration
 	@ConditionalOnClass(StreamsBuilder.class)
-	public static class KafkaStreamsConfiguration {
+	public static class KafkaStreamsAutoConfiguration {
 
 		@Bean(KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
-		public Properties defaultKafkaStreamsConfig(KafkaProperties properties) {
-			Properties streamsConfig = new Properties();
-			properties.buildStreamsProperties().forEach((k, v) -> {
-				String value = v.toString();
-				if (v instanceof List && value.length() > 1) {
-					// trim [...] - revert to comma-delimited list
-					value = value.substring(1, value.length() - 1);
+		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"));
 				}
-				streamsConfig.put(k, value);
-			});
-			return streamsConfig;
+			}
+			return new KafkaStreamsConfiguration(streamsProperties);
 		}
 
 		@Bean
-		public Object kafkaStreamsFactoryBeanConfigurer(
+		public KafkaStreamsFactoryBeanConfigurer kafkaStreamsFactoryBeanConfigurer(
 				StreamsBuilderFactoryBean factoryBean, KafkaProperties properties) {
 
-			factoryBean.setAutoStartup(properties.getStreams().isAutoStartup());
-			return null;
+			return new KafkaStreamsFactoryBeanConfigurer(factoryBean, properties);
 		}
 
 		@Configuration
 		@EnableKafkaStreams
-		public static class EnableKafkaStreamsConfiguration {
+		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());
+			}
 
 		}
 

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

@@ -31,7 +31,6 @@
 import org.apache.kafka.common.config.SslConfigs;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
-import org.apache.kafka.streams.StreamsConfig;
 
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.PropertyMapper;
@@ -58,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"));
@@ -231,7 +230,7 @@ public static class Consumer {
 
 		/**
 		 * 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;
 
@@ -441,7 +440,7 @@ public static class Producer {
 
 		/**
 		 * 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;
 
@@ -661,16 +660,16 @@ public static class Streams {
 		/**
 		 * Kafka streams application.id property; default spring.application.name.
 		 */
-		private String applicationId = "${spring.application.name}";
+		private String applicationId;
 
 		/**
-		 * Whether or not to auto-start the streams factory bean; default false.
+		 * 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.
+		 * connection to the Kafka cluster. Overrides the global property, for streams.
 		 */
 		private List<String> bootstrapServers;
 
@@ -767,16 +766,15 @@ public Map<String, String> getProperties() {
 		public Map<String, Object> buildProperties() {
 			Properties properties = new Properties();
 			PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
-			map.from(this::getApplicationId)
-					.to(properties.in(StreamsConfig.APPLICATION_ID_CONFIG));
+			map.from(this::getApplicationId).to(properties.in("application.id"));
 			map.from(this::getBootstrapServers)
-					.to(properties.in(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG));
+					.to(properties.in(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG));
 			map.from(this::getCacheMaxBytesBuffering)
-					.to(properties.in(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG));
-			map.from(this::getClientId).to(properties.in(StreamsConfig.CLIENT_ID_CONFIG));
-			map.from(this::getReplicationFactor)
-					.to(properties.in(StreamsConfig.REPLICATION_FACTOR_CONFIG));
-			map.from(this::getStateDir).to(properties.in(StreamsConfig.STATE_DIR_CONFIG));
+					.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);
 		}
 

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

@@ -42,6 +42,7 @@
 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;
@@ -297,7 +298,7 @@ public void streamsProperties() {
 				"spring.kafka.streams.ssl.protocol=TLSv1.2").run((context) -> {
 					Properties configs = context.getBean(
 							KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME,
-							Properties.class);
+							KafkaStreamsConfiguration.class).asProperties();
 					assertThat(configs.get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG))
 							.isEqualTo("localhost:9092, localhost:9093");
 					assertThat(

spring-boot-project/spring-boot-docs/src/main/asciidoc/appendix-application-properties.adoc

@@ -1048,11 +1048,11 @@ content into your application. Rather, pick only the properties that you need.
 	spring.kafka.admin.ssl.trust-store-location= # Location of the trust store file.
 	spring.kafka.admin.ssl.trust-store-password= # Store password for the trust store file.
 	spring.kafka.admin.ssl.trust-store-type= # Type of the trust store.
-	spring.kafka.bootstrap-servers= # Comma-delimited list of host:port pairs to use for establishing the initial connection to the Kafka cluster.
+	spring.kafka.bootstrap-servers= # Comma-delimited list of host:port pairs to use for establishing the initial connection to the Kafka cluster. Applies to all components unless overridden.
 	spring.kafka.client-id= # ID to pass to the server when making requests. Used for server-side logging.
 	spring.kafka.consumer.auto-commit-interval= # Frequency with which the consumer offsets are auto-committed to Kafka if 'enable.auto.commit' is set to true.
 	spring.kafka.consumer.auto-offset-reset= # What to do when there is no initial offset in Kafka or if the current offset no longer exists on the server.
-	spring.kafka.consumer.bootstrap-servers= # Comma-delimited list of host:port pairs to use for establishing the initial connection to the Kafka cluster.
+	spring.kafka.consumer.bootstrap-servers= # Comma-delimited list of host:port pairs to use for establishing the initial connection to the Kafka cluster. Overrides the global property, for consumers.
 	spring.kafka.consumer.client-id= # ID to pass to the server when making requests. Used for server-side logging.
 	spring.kafka.consumer.enable-auto-commit= # Whether the consumer's offset is periodically committed in the background.
 	spring.kafka.consumer.fetch-max-wait= # Maximum amount of time the server blocks before answering the fetch request if there isn't sufficient data to immediately satisfy the requirement given by "fetch.min.bytes".
@@ -1088,7 +1088,7 @@ content into your application. Rather, pick only the properties that you need.
 	spring.kafka.listener.type=single # Listener type.
 	spring.kafka.producer.acks= # Number of acknowledgments the producer requires the leader to have received before considering a request complete.
 	spring.kafka.producer.batch-size= # Default batch size in bytes.
-	spring.kafka.producer.bootstrap-servers= # Comma-delimited list of host:port pairs to use for establishing the initial connection to the Kafka cluster.
+	spring.kafka.producer.bootstrap-servers= # Comma-delimited list of host:port pairs to use for establishing the initial connection to the Kafka cluster. Overrides the global property, for producers.
 	spring.kafka.producer.buffer-memory= # Total bytes of memory the producer can use to buffer records waiting to be sent to the server.
 	spring.kafka.producer.client-id= # ID to pass to the server when making requests. Used for server-side logging.
 	spring.kafka.producer.compression-type= # Compression type for all data generated by the producer.
@@ -1114,7 +1114,8 @@ content into your application. Rather, pick only the properties that you need.
 	spring.kafka.ssl.trust-store-location= # Location of the trust store file.
 	spring.kafka.ssl.trust-store-password= # Store password for the trust store file.
 	spring.kafka.ssl.trust-store-type= # Type of the trust store.
-	spring.kafka.streams.auto-startup= # Whether or not to auto-start the streams factory bean; default false.
+	spring.kafka.streams.auto-startup= # Whether or not to auto-start the streams factory bean.
+	spring.kafka.streams.bootstrap-servers= # Comma-delimited list of host:port pairs to use for establishing the initial connection to the Kafka cluster. Overrides the global property, for streams.
 	spring.kafka.streams.cache-max-bytes-buffering= # Maximum number of memory bytes to be used for buffering across all threads.
 	spring.kafka.streams.client-id= # ID to pass to the server when making requests. Used for server-side logging.
 	spring.kafka.streams.properties.*= # Additional Kafka properties used to configure the streams.

spring-boot-project/spring-boot-docs/src/main/asciidoc/spring-boot-features.adoc

@@ -5632,10 +5632,11 @@ The following component creates a listener endpoint on the `someTopic` topic:
 Spring for Apache Kafka provides a factory bean to create a `StreamsBuilder` object and
 manage the lifecycle of its streams; the factory bean is created when
 `@EnableKafkaStreams` is present on a `@Configuration` class.
-The factory bean requires a `Properties` object for streams configuration.
+The factory bean requires a `KafkaStreamsConfiguration` object for streams configuration.
 
 If Spring Boot detects the `kafka-streams` jar on the classpath, it will auto-configure
-the `Properties` bean from the `KafkaProperties` object.
+the `KafkaStreamsConfiguration` bean from the `KafkaProperties` object as well as enabling
+the creation of the factory bean by spring-kafka.
 
 There are two required Kafka properties for streams (`bootstrap.servers` and
 `application.id`); by default, the `application.id` is set to the `spring.application.name`
@@ -5660,6 +5661,10 @@ public KStream<Integer, String> kStream(StreamsBuilder streamsBuilder) {
 ----
 ====
 
+By default, the factory bean `autoStartup` property is false; to automatically start the
+streams managed by the `StreamsBuilder` object it creates, set property
+`spting.kafka.streams.auto-startup=true`.
+
 
 [[boot-features-kafka-extra-props]]
 ==== Additional Kafka Properties