From 4c80027d28365c0e3897cd4ff6e642845f6521e5 Mon Sep 17 00:00:00 2001 From: wadhwaroh-lang Date: Sat, 11 Jul 2026 23:37:06 -0400 Subject: [PATCH 1/2] GH-3120 Align KafkaListenerContainerCustomizer with DLQ customizer defaults Make the three-argument configure() a no-op default and keep the four-argument method abstract so the interface can be implemented as a lambda that receives ExtendedConsumerProperties, matching ListenerContainerWithDlqAndRetryCustomizer. Signed-off-by: wadhwaroh-lang --- .../KafkaListenerContainerCustomizer.java | 19 +++++++++++--- ...KafkaListenerContainerCustomizerTests.java | 25 +++++++++++++++---- .../container-cust-kafka-binder.adoc | 12 ++++----- 3 files changed, 40 insertions(+), 16 deletions(-) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaListenerContainerCustomizer.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaListenerContainerCustomizer.java index 606ccd1d91..a8321b00e1 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaListenerContainerCustomizer.java +++ b/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> { + /** + * 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 extendedConsumerProperties) { - configure(container, destinationName, group); - } + void configure(AbstractMessageListenerContainer container, String destinationName, String group, + ExtendedConsumerProperties extendedConsumerProperties); } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaListenerContainerCustomizerTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaListenerContainerCustomizerTests.java index a6b87f7c78..b1d92c9c17 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaListenerContainerCustomizerTests.java +++ b/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> captured = new AtomicReference<>(); + KafkaListenerContainerCustomizer customizer = (container, destinationName, group, properties) -> + captured.set(properties); + + KafkaConsumerProperties kafkaConsumerProperties = new KafkaConsumerProperties(); + kafkaConsumerProperties.setEnableDlq(true); + ExtendedConsumerProperties 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) { - - } } } } diff --git a/docs/modules/ROOT/pages/kafka/kafka-binder/container-cust-kafka-binder.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/container-cust-kafka-binder.adoc index b1e6771afd..0f4316357b 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-binder/container-cust-kafka-binder.adoc +++ b/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> 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 extendedConsumerProperties) { - configure(container, destinationName, group); -} + ExtendedConsumerProperties extendedConsumerProperties); ---- This method extends the base `configure` method with an additional parameter: From 7be8b1f3d4ba6dc5cd059896c63cb8c338d309bb Mon Sep 17 00:00:00 2001 From: wadhwaroh-lang Date: Sat, 11 Jul 2026 23:40:15 -0400 Subject: [PATCH 2/2] Update CI PR workflow to Maven 3.9.16 Apache's Maven CDN no longer hosts 3.8.9, so jvalkeal/setup-maven fails before tests run. Point the workflow at the currently published 3.9.16. Signed-off-by: wadhwaroh-lang --- .github/workflows/ci-pr.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci-pr.yml b/.github/workflows/ci-pr.yml index 13a9bc009f..89e4407195 100644 --- a/.github/workflows/ci-pr.yml +++ b/.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