Add Kafka AutoConfiguration

Change Common Props to Hyphenated (asciidoc)

Only Support HIGH and a Few Other Props

Author: garyrussell

spring-boot-autoconfigure/pom.xml

@@ -472,6 +472,11 @@
 			<artifactId>spring-rabbit</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-spring-service-connector</artifactId>

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

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.autoconfigure.kafka;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.kafka.config.KafkaListenerConfigUtils;
+import org.springframework.kafka.config.KafkaListenerContainerFactory;
+import org.springframework.kafka.config.SimpleKafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
+
+/**
+ * Adds {@link EnableKafka} if present on the classpath.
+ *
+ * @author Gary Russell
+ * @since 1.4
+ *
+ */
+@Configuration
+@ConditionalOnClass(EnableKafka.class)
+class KafkaAnnotationDrivenConfiguration {
+
+	private final KafkaProperties properties;
+
+	KafkaAnnotationDrivenConfiguration(KafkaProperties properties) {
+		this.properties = properties;
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
+	public SimpleKafkaListenerContainerFactoryConfigurer kafkaListenerContainerFactoryConfigurer() {
+		SimpleKafkaListenerContainerFactoryConfigurer configurer = new SimpleKafkaListenerContainerFactoryConfigurer();
+		configurer.setKafkaProperties(this.properties);
+		return configurer;
+	}
+
+	@Bean
+	@ConditionalOnMissingBean(name = "kafkaListenerContainerFactory")
+	public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory(
+			SimpleKafkaListenerContainerFactoryConfigurer configurer,
+			ConsumerFactory<Object, Object> kafkaConsumerFactory) {
+		SimpleKafkaListenerContainerFactory<Object, Object> factory =
+				new SimpleKafkaListenerContainerFactory<Object, Object>();
+		configurer.configure(factory, kafkaConsumerFactory);
+		return factory;
+	}
+
+
+	@EnableKafka
+	@ConditionalOnMissingBean(name = KafkaListenerConfigUtils.KAFKA_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
+	protected static class EnableKafkaConfiguration {
+
+	}
+
+}

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

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.boot.autoconfigure.kafka;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+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.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.support.LoggingProducerListener;
+import org.springframework.kafka.support.ProducerListener;
+
+/**
+ * Auto-configuration for Spring for Apache Kafka.
+ *
+ * @author Gary Russell
+ * @since 1.4
+ *
+ */
+@Configuration
+@ConditionalOnClass(KafkaTemplate.class)
+@EnableConfigurationProperties(KafkaProperties.class)
+@Import(KafkaAnnotationDrivenConfiguration.class)
+public class KafkaAutoConfiguration {
+
+	private final KafkaProperties properties;
+
+	public KafkaAutoConfiguration(KafkaProperties properties) {
+		this.properties = properties;
+	}
+
+	@Bean
+	@ConditionalOnMissingBean(KafkaTemplate.class)
+	public KafkaTemplate<?, ?> kafkaTemplate(ProducerFactory<Object, Object> kafkaProducerFactory,
+			ProducerListener<Object, Object> kafkaProducerListener) {
+		KafkaTemplate<Object, Object> kafkaTemplate = new KafkaTemplate<Object, Object>(kafkaProducerFactory);
+		kafkaTemplate.setProducerListener(kafkaProducerListener);
+		kafkaTemplate.setDefaultTopic(this.properties.getTemplate().getDefaultTopic());
+		return kafkaTemplate;
+	}
+
+	@Bean
+	@ConditionalOnMissingBean(ProducerListener.class)
+	public ProducerListener<Object, Object> kafkaProducerListener() {
+		return new LoggingProducerListener<Object, Object>();
+	}
+
+	@Configuration
+	protected static class ConnectionConfig {
+
+		@Bean
+		@ConditionalOnMissingBean(ConsumerFactory.class)
+		public ConsumerFactory<?, ?> kafkaConsumerFactory(KafkaProperties properties) {
+			return new DefaultKafkaConsumerFactory<Object, Object>(properties.buildConsumerProperties());
+		}
+
+		@Bean
+		@ConditionalOnMissingBean(ProducerFactory.class)
+		public ProducerFactory<?, ?> kafkaProducerFactory(KafkaProperties properties) {
+			return new DefaultKafkaProducerFactory<Object, Object>(properties.buildProducerProperties());
+		}
+
+	}
+
+}