GH-1661: @KafkaListener Improvements

Resolves #1661

Allow properties that resolve to bean names to also resolve to instances
of the required type instead.

Author: garyrussell

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

@@ -104,7 +104,10 @@
 	/**
 	 * The bean name of the {@link org.springframework.kafka.config.KafkaListenerContainerFactory}
 	 * to use to create the message listener container responsible to serve this endpoint.
-	 * <p>If not specified, the default container factory is used, if any.
+	 * <p>
+	 * If not specified, the default container factory is used, if any. If a SpEL
+	 * expression is provided ({@code #{...}}), the expression can either evaluate to a
+	 * container factory instance or a bean name.
 	 * @return the container factory bean name.
 	 */
 	String containerFactory() default "";
@@ -156,7 +159,10 @@
 
 	/**
 	 * Set an {@link org.springframework.kafka.listener.KafkaListenerErrorHandler} bean
-	 * name to invoke if the listener method throws an exception.
+	 * name to invoke if the listener method throws an exception. If a SpEL expression is
+	 * provided ({@code #{...}}), the expression can either evaluate to a
+	 * {@link org.springframework.kafka.listener.KafkaListenerErrorHandler} instance or a
+	 * bean name.
 	 * @return the error handler.
 	 * @since 1.3
 	 */

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2020 the original author or authors.
+ * Copyright 2014-2021 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.
@@ -449,31 +449,75 @@ protected void processListener(MethodKafkaListenerEndpoint<?, ?> endpoint, Kafka
 		resolveKafkaProperties(endpoint, kafkaListener.properties());
 		endpoint.setSplitIterables(kafkaListener.splitIterables());
 
-		KafkaListenerContainerFactory<?> factory = null;
-		String containerFactoryBeanName = resolve(kafkaListener.containerFactory());
-		if (StringUtils.hasText(containerFactoryBeanName)) {
-			Assert.state(this.beanFactory != null, "BeanFactory must be set to obtain container factory by bean name");
-			try {
-				factory = this.beanFactory.getBean(containerFactoryBeanName, KafkaListenerContainerFactory.class);
-			}
-			catch (NoSuchBeanDefinitionException ex) {
-				throw new BeanInitializationException("Could not register Kafka listener endpoint on [" + adminTarget
-						+ "] for bean " + beanName + ", no " + KafkaListenerContainerFactory.class.getSimpleName()
-						+ " with id '" + containerFactoryBeanName + "' was found in the application context", ex);
-			}
+		String containerFactory = resolve(kafkaListener.containerFactory());
+		if (StringUtils.hasText(containerFactory)) {
+			this.registrar.registerEndpoint(endpoint, resolveContainerFactory(kafkaListener, containerFactory, beanName));
+		}
+		else {
+			this.registrar.registerEndpoint(endpoint);
 		}
 
 		endpoint.setBeanFactory(this.beanFactory);
 		String errorHandlerBeanName = resolveExpressionAsString(kafkaListener.errorHandler(), "errorHandler");
 		if (StringUtils.hasText(errorHandlerBeanName)) {
-			endpoint.setErrorHandler(this.beanFactory.getBean(errorHandlerBeanName, KafkaListenerErrorHandler.class));
+			resolveErrorHandler(endpoint, kafkaListener);
 		}
-		this.registrar.registerEndpoint(endpoint, factory);
 		if (StringUtils.hasText(beanRef)) {
 			this.listenerScope.removeListener(beanRef);
 		}
 	}
 
+	private void resolveErrorHandler(MethodKafkaListenerEndpoint<?, ?> endpoint, KafkaListener KafkaListener) {
+		Object errorHandler = resolveExpression(KafkaListener.errorHandler());
+		if (errorHandler instanceof KafkaListenerErrorHandler) {
+			endpoint.setErrorHandler((KafkaListenerErrorHandler) errorHandler);
+		}
+		else {
+			String errorHandlerBeanName = resolveExpressionAsString(KafkaListener.errorHandler(), "errorHandler");
+			if (StringUtils.hasText(errorHandlerBeanName)) {
+				endpoint.setErrorHandler(
+						this.beanFactory.getBean(errorHandlerBeanName, KafkaListenerErrorHandler.class));
+			}
+		}
+	}
+
+	@Nullable
+	private KafkaListenerContainerFactory<?> resolveContainerFactory(KafkaListener KafkaListener,
+			Object factoryTarget, String beanName) {
+
+		KafkaListenerContainerFactory<?> factory = null;
+		Object resolved = resolveExpression(KafkaListener.containerFactory());
+		if (resolved instanceof KafkaListenerContainerFactory) {
+			return (KafkaListenerContainerFactory<?>) resolved;
+		}
+		String containerFactoryBeanName = resolveExpressionAsString(KafkaListener.containerFactory(),
+				"containerFactory");
+		if (StringUtils.hasText(containerFactoryBeanName)) {
+			assertBeanFactory();
+			try {
+				factory = this.beanFactory.getBean(containerFactoryBeanName, KafkaListenerContainerFactory.class);
+			}
+			catch (NoSuchBeanDefinitionException ex) {
+				throw new BeanInitializationException(
+						noBeanFoundMessage(factoryTarget, beanName, containerFactoryBeanName,
+								KafkaListenerContainerFactory.class), ex);
+			}
+		}
+		return factory;
+	}
+
+	protected void assertBeanFactory() {
+		Assert.state(this.beanFactory != null, "BeanFactory must be set to obtain container factory by bean name");
+	}
+
+	protected String noBeanFoundMessage(Object target, String listenerBeanName, String requestedBeanName,
+			Class<?> expectedClass) {
+
+		return "Could not register Kafka listener endpoint on ["
+				+ target + "] for bean " + listenerBeanName + ", no '" + expectedClass.getSimpleName() + "' with id '"
+				+ requestedBeanName + "' was found in the application context";
+	}
+
 	private void resolveKafkaProperties(MethodKafkaListenerEndpoint<?, ?> endpoint, String[] propertyStrings) {
 		if (propertyStrings.length > 0) {
 			Properties properties = new Properties();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2020 the original author or authors.
+ * Copyright 2017-2021 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.
@@ -72,13 +72,19 @@ public class BatchListenerConversionTests {
 	@Autowired
 	private Config config;
 
+	@Autowired
+	private Listener listener1;
+
+	@Autowired
+	private Listener listener2;
+
 	@Autowired
 	private KafkaTemplate<Integer, Foo> template;
 
 	@Test
 	public void testBatchOfPojos() throws Exception {
-		doTest(this.config.listener1(), "blc1");
-		doTest(this.config.listener2(), "blc2");
+		doTest(this.listener1, "blc1");
+		doTest(this.listener2, "blc2");
 	}
 
 	private void doTest(Listener listener, String topic) throws InterruptedException {
@@ -173,13 +179,13 @@ public Map<String, Object> producerConfigs(EmbeddedKafkaBroker embeddedKafka) {
 		}
 
 		@Bean
-		public Listener listener1() {
-			return new Listener("blc1");
+		public Listener listener1(KafkaListenerContainerFactory<?> cf) {
+			return new Listener("blc1", cf);
 		}
 
 		@Bean
-		public Listener listener2() {
-			return new Listener("blc2");
+		public Listener listener2(KafkaListenerContainerFactory<?> cf) {
+			return new Listener("blc2", cf);
 		}
 
 		@Bean
@@ -202,17 +208,25 @@ public static class Listener {
 
 		private final CountDownLatch latch2 = new CountDownLatch(1);
 
+		private final KafkaListenerContainerFactory<?> cf;
+
 		private List<Foo> received;
 
 		private List<String> receivedTopics;
 
 		private List<Integer> receivedPartitions;
 
-		public Listener(String topic) {
+		public Listener(String topic, KafkaListenerContainerFactory<?> cf) {
 			this.topic = topic;
+			this.cf = cf;
+		}
+
+		public KafkaListenerContainerFactory<?> getContainerFactory() {
+			return this.cf;
 		}
 
-		@KafkaListener(topics = "#{__listener.topic}", groupId = "#{__listener.topic}.group")
+		@KafkaListener(topics = "#{__listener.topic}", groupId = "#{__listener.topic}.group",
+				containerFactory = "#{__listener.containerFactory}")
 		// @SendTo("foo") test WARN log for void return
 		public void listen1(List<Foo> foos, @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics,
 				@Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions) {