GH-2985: Add Kafka Listener Container Customizer interfaces and docum…

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

@@ -0,0 +1,46 @@
+/*
+ * 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 org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
+import org.springframework.cloud.stream.config.ListenerContainerCustomizer;
+import org.springframework.kafka.listener.AbstractMessageListenerContainer;
+
+/**
+ * Extension of {@link ListenerContainerCustomizer} specific to Kafka binder.
+ * This interface allows for customization of Kafka listener containers with
+ * access to Kafka-specific extended consumer properties.
+ *
+ * @author Soby Chacko
+ * @since 4.2.0
+ */
+public interface KafkaListenerContainerCustomizer extends ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> {
+
+	/**
+	 * Configure the Kafka listener container with access to extended consumer properties.
+	 *
+	 * @param container the Kafka message listener container to configure
+	 * @param destinationName the name of the destination (topic) that this listener container is associated with
+	 * @param group the consumer group name
+	 * @param extendedConsumerProperties the extended consumer properties specific to Kafka
+	 */
+	default void configure(AbstractMessageListenerContainer<?, ?> container, String destinationName, String group,
+						ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties) {
+		configure(container, destinationName, group);
+	}
+}

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

@@ -758,7 +758,10 @@ else if (!extendedConsumerProperties.isBatchMode() && transMan != null) {
 					? createBackOff(extendedConsumerProperties)
 					: null;
 			c.configure(messageListenerContainer, destination.getName(), consumerGroup, destinationResolver,
-							createBackOff);
+							createBackOff, extendedConsumerProperties);
+		}
+		else if (customizer instanceof KafkaListenerContainerCustomizer c) {
+			c.configure(messageListenerContainer, destination.getName(), consumerGroup, extendedConsumerProperties);
 		}
 		else {
 			((ListenerContainerCustomizer<Object>) customizer)

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2021 the original author or authors.
+ * Copyright 2021-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.
@@ -21,6 +21,8 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.TopicPartition;
 
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
 import org.springframework.cloud.stream.config.ListenerContainerCustomizer;
 import org.springframework.kafka.listener.AbstractMessageListenerContainer;
 import org.springframework.lang.Nullable;
@@ -31,12 +33,21 @@
  * metadata.
  *
  * @author Gary Russell
+ * @author Soby Chacko
  * @since 3.2
  *
  */
 public interface ListenerContainerWithDlqAndRetryCustomizer
 		extends ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> {
 
+	/**
+	 *
+	 * API method for configuring the container that also gives access to the {@link ExtendedConsumerProperties} for the binding.
+	 *
+	 * @param container the container.
+	 * @param destinationName the destination name.
+	 * @param group the consumer group.
+	 */
 	@Override
 	default void configure(AbstractMessageListenerContainer<?, ?> container, String destinationName, String group) {
 	}
@@ -55,6 +66,26 @@ void configure(AbstractMessageListenerContainer<?, ?> container, String destinat
 			@Nullable BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> dlqDestinationResolver,
 			@Nullable BackOff backOff);
 
+	/**
+	 *
+	 * API method for configuring the container that also gives access to the {@link ExtendedConsumerProperties} for the binding.
+	 *
+	 * @param container the container.
+	 * @param destinationName the destination name.
+	 * @param group the consumer group.
+	 * @param dlqDestinationResolver a destination resolver for the dead letter topic (if
+	 * enableDlq).
+	 * @param backOff the backOff using retry properties (if configured).
+	 * @param extendedConsumerProperties extended binding consumer properties.
+	 *
+	 * @since 4.2.0
+	 */
+	default void configure(AbstractMessageListenerContainer<?, ?> container, String destinationName, String group,
+						@Nullable BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> dlqDestinationResolver,
+						@Nullable BackOff backOff, ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties) {
+		configure(container, destinationName, group, dlqDestinationResolver, backOff);
+	}
+
 	/**
 	 * Return false to move retries and DLQ from the binding to a customized error handler
 	 * using the retry metadata and/or a {@code DeadLetterPublishingRecoverer} when

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

@@ -0,0 +1,162 @@
+/*
+ * 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.function.Consumer;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+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.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
+import org.springframework.kafka.listener.AbstractMessageListenerContainer;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.test.annotation.DirtiesContext;
+
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+
+/**
+ * @author Soby Chacko
+ * @since 4.2.0
+ */
+@SpringBootTest(
+	classes = {KafkaBinderConfiguration.class, KafkaListenerContainerCustomizerTests.TestConfig.class},
+	properties = {
+		"spring.cloud.function.definition=testConsumer",
+		"spring.cloud.stream.bindings.testConsumer-in-0.destination=test-topic",
+		"spring.cloud.stream.bindings.testConsumer-in-0.group=test-group",
+		"spring.cloud.stream.kafka.bindings.testConsumer-in-0.consumer.enableDlq=true"
+	}
+)
+@DirtiesContext
+@EmbeddedKafka(partitions = 1, topics = "test-topic")
+class KafkaListenerContainerCustomizerTests {
+
+	@Autowired
+	private KafkaListenerContainerCustomizer compositeCustomizer;
+
+	@Test
+	@SuppressWarnings("unchecked")
+	void customizersInvoked() {
+		KafkaListenerContainerCustomizer kafkaCustomizer =
+			((TestConfig.CompositeCustomizer) compositeCustomizer).getKafkaCustomizer();
+		ListenerContainerWithDlqAndRetryCustomizer dlqCustomizer =
+			((TestConfig.CompositeCustomizer) compositeCustomizer).getDlqCustomizer();
+
+		ArgumentCaptor<ExtendedConsumerProperties<KafkaConsumerProperties>> kafkaPropertiesCaptor = ArgumentCaptor.forClass(ExtendedConsumerProperties.class);
+		ArgumentCaptor<ExtendedConsumerProperties<KafkaConsumerProperties>> dlqPropertiesCaptor = ArgumentCaptor.forClass(ExtendedConsumerProperties.class);
+
+		verify(kafkaCustomizer, timeout(5000).times(1)).configure(
+			any(AbstractMessageListenerContainer.class),
+			eq("test-topic"),
+			eq("test-group"),
+			kafkaPropertiesCaptor.capture()
+		);
+
+		verify(dlqCustomizer, timeout(5000).times(1)).configure(
+			any(AbstractMessageListenerContainer.class),
+			eq("test-topic"),
+			eq("test-group"),
+			isNull(),
+			isNull(),
+			dlqPropertiesCaptor.capture()
+		);
+
+		ExtendedConsumerProperties<KafkaConsumerProperties> kafkaProperties = kafkaPropertiesCaptor.getValue();
+		ExtendedConsumerProperties<KafkaConsumerProperties> dlqProperties = dlqPropertiesCaptor.getValue();
+
+		// Assert common properties
+		assertThat(kafkaProperties.getBindingName()).isEqualTo("testConsumer-in-0");
+		assertThat(dlqProperties.getBindingName()).isEqualTo("testConsumer-in-0");
+
+		// Assert Kafka-specific properties
+		assertThat(kafkaProperties.getExtension())
+			.satisfies(extension -> {
+				assertThat(extension.isEnableDlq()).isTrue();
+				assertThat(extension.isAutoRebalanceEnabled()).isTrue();
+			});
+
+		// Assert that both captured properties are the same instance
+		assertThat(kafkaProperties).isSameAs(dlqProperties);
+	}
+
+	@Configuration
+	@EnableAutoConfiguration
+	static class TestConfig {
+
+		@Bean
+		public Consumer<String> testConsumer() {
+			return message -> {
+				// Do nothing, just to trigger consumer binding
+			};
+		}
+
+		@Bean
+		@Primary
+		public KafkaListenerContainerCustomizer compositeCustomizer() {
+			return new CompositeCustomizer(
+				mock(KafkaListenerContainerCustomizer.class),
+				mock(ListenerContainerWithDlqAndRetryCustomizer.class)
+			);
+		}
+
+		static class CompositeCustomizer implements KafkaListenerContainerCustomizer {
+			private final KafkaListenerContainerCustomizer kafkaCustomizer;
+			private final ListenerContainerWithDlqAndRetryCustomizer dlqCustomizer;
+
+			CompositeCustomizer(KafkaListenerContainerCustomizer kafkaCustomizer,
+				ListenerContainerWithDlqAndRetryCustomizer dlqCustomizer) {
+				this.kafkaCustomizer = kafkaCustomizer;
+				this.dlqCustomizer = dlqCustomizer;
+			}
+
+			@Override
+			public void configure(AbstractMessageListenerContainer<?, ?> container, String destinationName,
+				String group, ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties) {
+				kafkaCustomizer.configure(container, destinationName, group, extendedConsumerProperties);
+				dlqCustomizer.configure(container, destinationName, group, null, null, extendedConsumerProperties);
+			}
+
+			public KafkaListenerContainerCustomizer getKafkaCustomizer() {
+				return kafkaCustomizer;
+			}
+
+			public ListenerContainerWithDlqAndRetryCustomizer getDlqCustomizer() {
+				return dlqCustomizer;
+			}
+
+			@Override
+			public void configure(AbstractMessageListenerContainer<?, ?> container, String destinationName, String group) {
+
+			}
+		}
+	}
+}

docs/modules/ROOT/nav.adoc

@@ -58,6 +58,7 @@
 **** xref:kafka/kafka-binder/tombstone.adoc[]
 **** xref:kafka/kafka-binder/rebalance_listener.adoc[]
 **** xref:kafka/kafka-binder/retry-dlq.adoc[]
+**** xref:kafka/kafka-binder/container-cust-kafka-binder.adoc[]
 **** xref:kafka/kafka-binder/cons-prod-config-cust.adoc[]
 **** xref:kafka/kafka-binder/admin-client-config-cust.adoc[]
 **** xref:kafka/kafka-binder/custom-health-ind.adoc[]