GH-1872: @KafkaListener Override Listener Type

Resolves #1872

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -1524,6 +1524,9 @@ public KafkaListenerContainerFactory<?, ?> batchFactory() {
 ----
 ====
 
+NOTE: Starting with version 2.8, you can override the factory's `batchListener` propery using the `batch` property on the `@KafkaListener` annotation.
+This, together with the changes to <<error-handlers>> allows the same factory to be used for both record and batch listeners.
+
 The following example shows how to receive a list of payloads:
 
 ====

spring-kafka-docs/src/main/asciidoc/whats-new.adoc

@@ -15,6 +15,13 @@ The listener container can now be configured to accept manual offset commits out
 The container will defer the commit until the missing offset is acknowledged.
 See <<ooo-commits>> for more information.
 
+[[x28-batch-overrude]]
+==== `@KafkaListener` Changes
+
+It is now possible to specify whether the listener method is a batch listener on the method itself.
+This allows the same container factory to be used for both record and batch listeners.
+See <<batch-listeners>> for more information.
+
 [[x28-template]]
 ==== `KafkaTemplate` Changes
 

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

@@ -283,4 +283,18 @@
 	 */
 	String contentTypeConverter() default "";
 
+	/**
+	 * Override the container factory's {@code batchListener} property. The listener
+	 * method signature should receive a {@code List<?>}; refer to the reference
+	 * documentation. This allows a single container factory to be used for both record
+	 * and batch listeners; previously separate container factories were required.
+	 * @return "true" for the annotated method to be a batch listener or "false" for a
+	 * record listener. If not set, the container factory setting is used. SpEL and
+	 * property placeholders are not supported because the listener type cannot be
+	 * variable.
+	 * @since 2.8
+	 * @see Boolean#parseBoolean(String)
+	 */
+	String batch() default "";
+
 }

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

@@ -560,12 +560,16 @@ protected void processListener(MethodKafkaListenerEndpoint<?, ?> endpoint, Kafka
 		}
 	}
 
-	private void processKafkaListenerAnnotationForRetryTopic(MethodKafkaListenerEndpoint<?, ?> endpoint, KafkaListener kafkaListener, Object bean) {
+	private void processKafkaListenerAnnotationForRetryTopic(MethodKafkaListenerEndpoint<?, ?> endpoint,
+			KafkaListener kafkaListener, Object bean) {
+
 		processKafkaListenerAnnotationBeforeRegistration(endpoint, kafkaListener, bean);
 		processKafkaListenerEndpointAfterRegistration(endpoint, kafkaListener);
 	}
 
-	private void processKafkaListenerAnnotationBeforeRegistration(MethodKafkaListenerEndpoint<?, ?> endpoint, KafkaListener kafkaListener, Object bean) {
+	private void processKafkaListenerAnnotationBeforeRegistration(MethodKafkaListenerEndpoint<?, ?> endpoint,
+			KafkaListener kafkaListener, Object bean) {
+
 		endpoint.setBean(bean);
 		endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);
 		endpoint.setId(getEndpointId(kafkaListener));
@@ -591,9 +595,14 @@ private void processKafkaListenerAnnotationBeforeRegistration(MethodKafkaListene
 		}
 		resolveKafkaProperties(endpoint, kafkaListener.properties());
 		endpoint.setSplitIterables(kafkaListener.splitIterables());
+		if (StringUtils.hasText(kafkaListener.batch())) {
+			endpoint.setBatchListener(Boolean.parseBoolean(kafkaListener.batch()));
+		}
 	}
 
-	private void processKafkaListenerEndpointAfterRegistration(MethodKafkaListenerEndpoint<?, ?> endpoint, KafkaListener kafkaListener) {
+	private void processKafkaListenerEndpointAfterRegistration(MethodKafkaListenerEndpoint<?, ?> endpoint,
+			KafkaListener kafkaListener) {
+
 		endpoint.setBeanFactory(this.beanFactory);
 		String errorHandlerBeanName = resolveExpressionAsString(kafkaListener.errorHandler(), "errorHandler");
 		if (StringUtils.hasText(errorHandlerBeanName)) {

spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java

@@ -395,10 +395,14 @@ private void configureEndpoint(AbstractKafkaListenerEndpoint<K, V> aklEndpoint)
 				.acceptIfNotNull(this.retryTemplate, aklEndpoint::setRetryTemplate)
 				.acceptIfNotNull(this.recoveryCallback, aklEndpoint::setRecoveryCallback)
 				.acceptIfNotNull(this.statefulRetry, aklEndpoint::setStatefulRetry)
-				.acceptIfNotNull(this.batchListener, aklEndpoint::setBatchListener)
 				.acceptIfNotNull(this.replyTemplate, aklEndpoint::setReplyTemplate)
 				.acceptIfNotNull(this.replyHeadersConfigurer, aklEndpoint::setReplyHeadersConfigurer)
 				.acceptIfNotNull(this.batchToRecordAdapter, aklEndpoint::setBatchToRecordAdapter);
+		if (aklEndpoint.getBatchListener() == null) {
+			JavaUtils.INSTANCE
+					.acceptIfNotNull(this.batchListener, aklEndpoint::setBatchListener);
+
+		}
 	}
 
 	/**

spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java

@@ -100,7 +100,7 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 
 	private boolean statefulRetry;
 
-	private boolean batchListener;
+	private Boolean batchListener;
 
 	private KafkaTemplate<?, ?> replyTemplate;
 
@@ -263,6 +263,17 @@ public void setGroup(String group) {
 	 * @since 1.1
 	 */
 	public boolean isBatchListener() {
+		return this.batchListener == null ? false : this.batchListener;
+	}
+
+	/**
+	 * Return the current batch listener flag for this endpoint, or null if not explicitly
+	 * set.
+	 * @return the batch listener flag.
+	 * @since 2.8
+	 */
+	@Nullable
+	public Boolean getBatchListener() {
 		return this.batchListener;
 	}
 
@@ -509,9 +520,10 @@ private void setupMessageListener(MessageListenerContainer container,
 		}
 		adapter.setSplitIterables(this.splitIterables);
 		Object messageListener = adapter;
+		boolean isBatchListener = isBatchListener();
 		Assert.state(messageListener != null,
 				() -> "Endpoint [" + this + "] must provide a non null message listener");
-		Assert.state(this.retryTemplate == null || !this.batchListener,
+		Assert.state(this.retryTemplate == null || !isBatchListener,
 				"A 'RetryTemplate' is not supported with a batch listener; consider configuring the container "
 				+ "with a suitably configured 'SeekToCurrentBatchErrorHandler' instead");
 		if (this.retryTemplate != null) {
@@ -520,9 +532,9 @@ private void setupMessageListener(MessageListenerContainer container,
 					this.retryTemplate, this.recoveryCallback, this.statefulRetry);
 		}
 		if (this.recordFilterStrategy != null) {
-			if (this.batchListener) {
+			if (isBatchListener) {
 				if (((MessagingMessageListenerAdapter<K, V>) messageListener).isConsumerRecords()) {
-					this.logger.warn(() -> "Filter strategy ignored when consuming 'ConsumerRecords'"
+					this.logger.warn(() -> "Filter strategy ignored when consuming 'ConsumerRecords' instead of a List"
 							+ (this.id != null ? " id: " + this.id : ""));
 				}
 				else {

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

@@ -27,6 +27,8 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 
+import java.io.PrintWriter;
+import java.io.StringWriter;
 import java.lang.reflect.Type;
 import java.time.Duration;
 import java.util.Collection;
@@ -104,6 +106,7 @@
 import org.springframework.kafka.listener.ConsumerSeekAware;
 import org.springframework.kafka.listener.ContainerProperties;
 import org.springframework.kafka.listener.ContainerProperties.AckMode;
+import org.springframework.kafka.listener.DefaultErrorHandler;
 import org.springframework.kafka.listener.KafkaListenerErrorHandler;
 import org.springframework.kafka.listener.KafkaMessageListenerContainer;
 import org.springframework.kafka.listener.ListenerExecutionFailedException;
@@ -261,9 +264,12 @@ public void testAnonymous() {
 	public void testSimple() throws Exception {
 		this.recordFilter.called = false;
 		template.send("annotated1", 0, "foo");
+		template.send("annotated1", 0, "bar");
 		assertThat(this.listener.latch1.await(60, TimeUnit.SECONDS)).isTrue();
 		assertThat(this.config.globalErrorThrowable).isNotNull();
 		assertThat(this.listener.receivedGroupId).isEqualTo("foo");
+		assertThat(this.listener.latch1Batch.await(60, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.listener.batchOverrideStackTrace).contains("BatchMessagingMessageListenerAdapter");
 
 		template.send("annotated2", 0, 123, "foo");
 		assertThat(this.listener.latch2.await(60, TimeUnit.SECONDS)).isTrue();
@@ -1010,11 +1016,20 @@ public PlatformTransactionManager transactionManager() {
 			factory.setConsumerFactory(consumerFactory());
 			factory.setRecordFilterStrategy(recordFilter());
 			factory.setReplyTemplate(partitionZeroReplyTemplate());
-			factory.setErrorHandler((ConsumerAwareErrorHandler) (t, d, c) -> {
-				this.globalErrorThrowable = t;
-				c.seek(new org.apache.kafka.common.TopicPartition(d.topic(), d.partition()), d.offset());
+			factory.setCommonErrorHandler(new DefaultErrorHandler() {
+
+				@Override
+				public void handleRemaining(Exception thrownException, List<ConsumerRecord<?, ?>> records,
+						Consumer<?, ?> consumer, MessageListenerContainer container) {
+
+					Config.this.globalErrorThrowable = thrownException;
+					super.handleRemaining(thrownException, records, consumer, container);
+				}
+
 			});
 			factory.getContainerProperties().setMicrometerTags(Collections.singletonMap("extraTag", "foo"));
+			// ensure annotation wins, even with explicitly set here
+			factory.setBatchListener(false);
 			return factory;
 		}
 
@@ -1671,7 +1686,9 @@ static class Listener implements ConsumerSeekAware {
 
 		private final ThreadLocal<ConsumerSeekCallback> seekCallBack = new ThreadLocal<>();
 
-		final CountDownLatch latch1 = new CountDownLatch(1);
+		final CountDownLatch latch1 = new CountDownLatch(2);
+
+		final CountDownLatch latch1Batch = new CountDownLatch(2);
 
 		final CountDownLatch latch2 = new CountDownLatch(2); // seek
 
@@ -1783,6 +1800,8 @@ static class Listener implements ConsumerSeekAware {
 
 		volatile Foo contentFoo;
 
+		volatile String batchOverrideStackTrace;
+
 		@KafkaListener(id = "manualStart", topics = "manualStart",
 				containerFactory = "kafkaAutoStartFalseListenerContainerFactory",
 				properties = { ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG + ":301000",
@@ -1801,6 +1820,15 @@ public void listen1(String foo, @Header(value = KafkaHeaders.GROUP_ID, required
 			this.latch1.countDown();
 		}
 
+		@KafkaListener(id = "fooBatch", topics = "annotated1", batch = "true")
+		public void listen1Batch(List<String> foo) {
+			StringWriter stringWriter = new StringWriter();
+			PrintWriter printWriter = new PrintWriter(stringWriter, true);
+			new RuntimeException().printStackTrace(printWriter);
+			this.batchOverrideStackTrace = stringWriter.getBuffer().toString();
+			foo.forEach(s ->  this.latch1Batch.countDown());
+		}
+
 		@KafkaListener(id = "bar", topicPattern = "${topicTwo:annotated2}")
 		public void listen2(@Payload String foo,
 				@Header(KafkaHeaders.GROUP_ID) String groupId,