GH-3120 Align KafkaListenerContainerCustomizer with DLQ customizer de…

.github/workflows/ci-pr.yml

@@ -38,7 +38,7 @@ jobs:
       - name: Setup Maven
         uses: jvalkeal/setup-maven@v1
         with:
-          maven-version: 3.8.9
+          maven-version: 3.9.16
           maven-mirror: 'https://dlcdn.apache.org/maven/maven-3/'
 
       - name: Build and run unit tests

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

@@ -31,6 +31,19 @@
  */
 public interface KafkaListenerContainerCustomizer extends ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> {
 
+	/**
+	 * No-op override of {@link ListenerContainerCustomizer#configure(Object, String, String)}.
+	 * Implementations should override
+	 * {@link #configure(AbstractMessageListenerContainer, String, String, ExtendedConsumerProperties)}
+	 * instead, which is the method invoked by the Kafka binder.
+	 * @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
+	 */
+	@Override
+	default void configure(AbstractMessageListenerContainer<?, ?> container, String destinationName, String group) {
+	}
+
 	/**
 	 * Configure the Kafka listener container with access to extended consumer properties.
 	 *
@@ -39,8 +52,6 @@ public interface KafkaListenerContainerCustomizer extends ListenerContainerCusto
 	 * @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);
-	}
+	void configure(AbstractMessageListenerContainer<?, ?> container, String destinationName, String group,
+			ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties);
 }

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

@@ -16,6 +16,7 @@
 
 package org.springframework.cloud.stream.binder.kafka;
 
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 
 import org.junit.jupiter.api.Test;
@@ -63,6 +64,25 @@ class KafkaListenerContainerCustomizerTests {
 	@Autowired
 	private KafkaListenerContainerCustomizer compositeCustomizer;
 
+	@Test
+	@SuppressWarnings("unchecked")
+	void canBeImplementedAsLambdaWithExtendedProperties() {
+		AtomicReference<ExtendedConsumerProperties<KafkaConsumerProperties>> captured = new AtomicReference<>();
+		KafkaListenerContainerCustomizer customizer = (container, destinationName, group, properties) ->
+			captured.set(properties);
+
+		KafkaConsumerProperties kafkaConsumerProperties = new KafkaConsumerProperties();
+		kafkaConsumerProperties.setEnableDlq(true);
+		ExtendedConsumerProperties<KafkaConsumerProperties> properties =
+			new ExtendedConsumerProperties<>(kafkaConsumerProperties);
+		AbstractMessageListenerContainer<?, ?> container = mock(AbstractMessageListenerContainer.class);
+
+		customizer.configure(container, "topic", "group", properties);
+
+		assertThat(captured.get()).isSameAs(properties);
+		assertThat(captured.get().getExtension().isEnableDlq()).isTrue();
+	}
+
 	@Test
 	@SuppressWarnings("unchecked")
 	void customizersInvoked() {
@@ -152,11 +172,6 @@ public KafkaListenerContainerCustomizer getKafkaCustomizer() {
 			public ListenerContainerWithDlqAndRetryCustomizer getDlqCustomizer() {
 				return dlqCustomizer;
 			}
-
-			@Override
-			public void configure(AbstractMessageListenerContainer<?, ?> container, String destinationName, String group) {
-
-			}
 		}
 	}
 }

docs/modules/ROOT/pages/kafka/kafka-binder/container-cust-kafka-binder.adoc

@@ -51,24 +51,22 @@ To use the `KafkaListenerContainerCustomizer`, create a bean that implements thi
 [source,java]
 ----
 @Bean
-public KafkaListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> kafkaCustomizer() {
+public KafkaListenerContainerCustomizer kafkaCustomizer() {
     return (container, destinationName, group, properties) -> {
         // Customize the Kafka container here
     };
 }
 ----
 
-The `KafkaListenerContainerCustomizer` interface adds the following method:
+The `KafkaListenerContainerCustomizer` interface makes the base three-argument `configure` method a no-op default (so it can be implemented as a lambda) and adds the following method:
 
 [source,java]
 ----
-default void configureKafkaListenerContainer(
-    C container,
+void configure(
+    AbstractMessageListenerContainer<?, ?> container,
     String destinationName,
     String group,
-    ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties) {
-        configure(container, destinationName, group);
-}
+    ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties);
 ----
 
 This method extends the base `configure` method with an additional parameter: