Avoid BeanDefinitionOverrideEx with EnableKafka

The current `KafkaBootstrapConfiguration` provides a couple beans
which can be overridden in the target project, but at the same time
the latest Spring Boot is going to fail with the
`BeanDefinitionOverrideException`

* Change `KafkaBootstrapConfiguration` to be as an
`ImportBeanDefinitionRegistrar` instead of `@Configuration`
* Check for bean definition existence before registering
`KafkaListenerAnnotationBeanPostProcessor` and `KafkaListenerEndpointRegistry`
in this `KafkaBootstrapConfiguration`
* Deffer such a bean definition processing with the
`KafkaListenerConfigurationSelector`

Author: artembilan

spring-kafka/src/main/java/org/springframework/kafka/annotation/EnableKafka.java

@@ -241,6 +241,7 @@
  *
  * @author Stephane Nicoll
  * @author Gary Russell
+ * @author Artem Bilan
  *
  * @see KafkaListener
  * @see KafkaListenerAnnotationBeanPostProcessor
@@ -250,6 +251,6 @@
 @Target(ElementType.TYPE)
 @Retention(RetentionPolicy.RUNTIME)
 @Documented
-@Import(KafkaBootstrapConfiguration.class)
+@Import(KafkaListenerConfigurationSelector.class)
 public @interface EnableKafka {
 }

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaBootstrapConfiguration.java

@@ -16,15 +16,15 @@
 
 package org.springframework.kafka.annotation;
 
-import org.springframework.beans.factory.config.BeanDefinition;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Role;
+import org.springframework.beans.factory.support.BeanDefinitionRegistry;
+import org.springframework.beans.factory.support.RootBeanDefinition;
+import org.springframework.context.annotation.ImportBeanDefinitionRegistrar;
+import org.springframework.core.type.AnnotationMetadata;
 import org.springframework.kafka.config.KafkaListenerConfigUtils;
 import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
 
 /**
- * {@code @Configuration} class that registers a {@link KafkaListenerAnnotationBeanPostProcessor}
+ * An {@link ImportBeanDefinitionRegistrar} class that registers a {@link KafkaListenerAnnotationBeanPostProcessor}
  * bean capable of processing Spring's @{@link KafkaListener} annotation. Also register
  * a default {@link KafkaListenerEndpointRegistry}.
  *
@@ -39,18 +39,21 @@
  * @see KafkaListenerEndpointRegistry
  * @see EnableKafka
  */
-@Configuration(proxyBeanMethods = false)
-public class KafkaBootstrapConfiguration {
+public class KafkaBootstrapConfiguration implements ImportBeanDefinitionRegistrar {
 
-	@Bean(name = KafkaListenerConfigUtils.KAFKA_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
-	@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
-	public KafkaListenerAnnotationBeanPostProcessor<?, ?> kafkaListenerAnnotationProcessor() {
-		return new KafkaListenerAnnotationBeanPostProcessor<>();
-	}
+	@Override
+	public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
+		if (!registry.containsBeanDefinition(
+				KafkaListenerConfigUtils.KAFKA_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)) {
+
+			registry.registerBeanDefinition(KafkaListenerConfigUtils.KAFKA_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME,
+					new RootBeanDefinition(KafkaListenerAnnotationBeanPostProcessor.class));
+		}
 
-	@Bean(name = KafkaListenerConfigUtils.KAFKA_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)
-	public KafkaListenerEndpointRegistry defaultKafkaListenerEndpointRegistry() {
-		return new KafkaListenerEndpointRegistry();
+		if (!registry.containsBeanDefinition(KafkaListenerConfigUtils.KAFKA_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)) {
+			registry.registerBeanDefinition(KafkaListenerConfigUtils.KAFKA_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME,
+					new RootBeanDefinition(KafkaListenerEndpointRegistry.class));
+		}
 	}
 
 }

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerConfigurationSelector.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.kafka.annotation;
+
+import org.springframework.context.annotation.DeferredImportSelector;
+import org.springframework.core.annotation.Order;
+import org.springframework.core.type.AnnotationMetadata;
+
+/**
+ * A {@link DeferredImportSelector} implementation with the lowest order to import a
+ * {@link KafkaBootstrapConfiguration} as late as possible.
+ *
+ * @author Artem Bilan
+ *
+ * @since 2.3
+ */
+@Order
+public class KafkaListenerConfigurationSelector implements DeferredImportSelector {
+
+	@Override
+	public String[] selectImports(AnnotationMetadata importingClassMetadata) {
+		return new String[] { KafkaBootstrapConfiguration.class.getName() };
+	}
+
+}

spring-kafka/src/test/java/org/springframework/kafka/annotation/AliasPropertiesTests.java

@@ -34,7 +34,9 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.annotation.AliasFor;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.config.KafkaListenerConfigUtils;
 import org.springframework.kafka.config.KafkaListenerContainerFactory;
+import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
@@ -46,6 +48,8 @@
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.2
  *
  */
@@ -59,10 +63,14 @@ public class AliasPropertiesTests {
 	@Autowired
 	private Config config;
 
+	@Autowired
+	private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;
+
 	@Test
 	public void testAliasFor() throws Exception {
 		this.template.send("alias.tests", "foo");
 		assertThat(this.config.latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.config.kafkaListenerEndpointRegistry()).isSameAs(this.kafkaListenerEndpointRegistry);
 	}
 
 	@Configuration
@@ -71,6 +79,11 @@ public static class Config {
 
 		private final CountDownLatch latch = new CountDownLatch(1);
 
+		@Bean(KafkaListenerConfigUtils.KAFKA_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)
+		public KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry() {
+			return new KafkaListenerEndpointRegistry();
+		}
+
 		@Bean
 		public EmbeddedKafkaBroker embeddedKafka() {
 			return new EmbeddedKafkaBroker(1, true, "alias.tests");