GH-2951: Batch consumer and DLQ issues

Batch consumer with `ListenerContainerWithDlqAndRetryCustomizer` should disable binder-based DLQ.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2951

* Addressing PR review comments

* Addressing review

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2023 the original author or authors.
+ * Copyright 2014-2024 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.
@@ -1077,6 +1077,11 @@ public class KafkaMessageChannelBinder extends
 		return new RawRecordHeaderErrorMessageStrategy();
 	}
 
+	private Boolean isBatchAndListenerContainerWithDlqAndRetryCustomizer(ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
+		ListenerContainerCustomizer<?> customizer = getContainerCustomizer();
+		return properties.isBatchMode() && customizer instanceof ListenerContainerWithDlqAndRetryCustomizer;
+	}
+
 	@SuppressWarnings("unchecked")
 	@Override
 	protected MessageHandler getErrorMessageHandler(final ConsumerDestination destination,
@@ -1084,7 +1089,7 @@ public class KafkaMessageChannelBinder extends
 			final ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
 
 		KafkaConsumerProperties kafkaConsumerProperties = properties.getExtension();
-		if (kafkaConsumerProperties.isEnableDlq()) {
+		if (kafkaConsumerProperties.isEnableDlq() && !isBatchAndListenerContainerWithDlqAndRetryCustomizer(properties)) {
 			KafkaProducerProperties dlqProducerProperties = kafkaConsumerProperties
 					.getDlqProducerProperties();
 			KafkaAwareTransactionManager<byte[], byte[]> transMan = transactionManager(

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/BatchWithDlqCustomizerDisablesBinderDlqTests.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2024-2024 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.cloud.stream.binder.kafka;
+
+import java.util.List;
+import java.util.function.Consumer;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.stream.binder.DefaultBinderFactory;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.test.annotation.DirtiesContext;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Soby Chacko
+ */
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = {
+	"spring.cloud.function.definition=consumer1",
+	"spring.cloud.stream.bindings.consumer-in-0.group=batchWithDlqCustomizerDisablesBinderDlq",
+	"spring.cloud.stream.kafka.bindings.consumer1-in-0.consumer.enable-dlq=true"})
+@EmbeddedKafka
+@DirtiesContext
+public class BatchWithDlqCustomizerDisablesBinderDlqTests {
+
+	@Autowired
+	private DefaultBinderFactory binderFactory;
+
+	@Test
+	void batchWithDlqCustomizerDisablesBinderDlq() {
+		KafkaMessageChannelBinder kafka = (KafkaMessageChannelBinder) this.binderFactory.getBinder("kafka", MessageChannel.class);
+
+		KafkaConsumerProperties kafkaConsumerProperties =
+			kafka.getExtendedConsumerProperties("consumer1-in-0");
+		ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties =
+			new ExtendedConsumerProperties<>(kafkaConsumerProperties);
+		extendedConsumerProperties.setBatchMode(true);
+
+		MessageHandler errorMessageHandler =
+			kafka.getErrorMessageHandler(null, null, extendedConsumerProperties);
+		// verifies that binder does not create a message handler for errors, which otherwise creates a handler for DLQ.
+		assertThat(errorMessageHandler).isNull();
+	}
+
+	@EnableAutoConfiguration
+	@Configuration
+	public static class BatchWithDlqDisablesBinderDlqTestsConfig {
+
+		@Bean
+		Consumer<Message<List<String>>> consumer1() {
+			return message -> {
+			};
+		}
+
+		@Bean
+		ListenerContainerWithDlqAndRetryCustomizer customizer() {
+			return (container, destinationName, group, dlqDestinationResolver, backOff) -> {
+			};
+		}
+	}
+
+}