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 new file mode 100644 index 000000000..bda6c8151 --- /dev/null +++ b/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> { + + /** + * 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 extendedConsumerProperties) { + configure(container, destinationName, group); + } +} diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 0b4d6f29b..f68576438 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -758,7 +758,10 @@ public class KafkaMessageChannelBinder extends ? 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) customizer) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/ListenerContainerWithDlqAndRetryCustomizer.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/ListenerContainerWithDlqAndRetryCustomizer.java index ed453c097..7f6c9d102 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/ListenerContainerWithDlqAndRetryCustomizer.java +++ b/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 java.util.function.BiFunction; 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 @@ import org.springframework.util.backoff.BackOff; * metadata. * * @author Gary Russell + * @author Soby Chacko * @since 3.2 * */ public interface ListenerContainerWithDlqAndRetryCustomizer extends ListenerContainerCustomizer> { + /** + * + * 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 @@ public interface ListenerContainerWithDlqAndRetryCustomizer @Nullable BiFunction, 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, Exception, TopicPartition> dlqDestinationResolver, + @Nullable BackOff backOff, ExtendedConsumerProperties 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 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 new file mode 100644 index 000000000..cf5c89f88 --- /dev/null +++ b/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> kafkaPropertiesCaptor = ArgumentCaptor.forClass(ExtendedConsumerProperties.class); + ArgumentCaptor> 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 kafkaProperties = kafkaPropertiesCaptor.getValue(); + ExtendedConsumerProperties 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 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 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) { + + } + } + } +} diff --git a/docs/modules/ROOT/nav.adoc b/docs/modules/ROOT/nav.adoc index 45549646b..f5ea867ff 100644 --- a/docs/modules/ROOT/nav.adoc +++ b/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[] 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 new file mode 100644 index 000000000..b1e6771af --- /dev/null +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/container-cust-kafka-binder.adoc @@ -0,0 +1,127 @@ += Kafka Binder Listener Container Customizers + +Spring Cloud Stream provides powerful customization options for message listener containers through the use of customizers. +This section covers the customizer interfaces available for Kafka: `ListenerContainerCustomizer`, its Kafka-specific extension `KafkaListenerContainerCustomizer`, and the specialized `ListenerContainerWithDlqAndRetryCustomizer`. + +== ListenerContainerCustomizer + +The `ListenerContainerCustomizer` is a generic interface in Spring Cloud Stream that allows customization of message listener containers. + +=== Purpose + +Use this customizer when you need to modify the behavior of the listener container. + +=== Usage + +To use the `ListenerContainerCustomizer`, create a bean that implements this interface in your configuration: + +[source,java] +---- +@Bean +public ListenerContainerCustomizer> genericCustomizer() { + return (container, destinationName, group) -> { + // Customize the container here + }; +} +---- + +The `ListenerContainerCustomizer` interface defines the following method: + +[source,java] +---- +void configure(C container, String destinationName, String group); +---- + +* `container`: The message listener container to customize. +* `destinationName`: The name of the destination (topic). +* `group`: The consumer group ID. + +== KafkaListenerContainerCustomizer + +The `KafkaListenerContainerCustomizer` interface extends `ListenerContainerCustomizer` to modify the behavior of the listener container and provides access to the binding-specific extended Kafka consumer properties. + +=== Purpose + +Use this customizer when you need to access the binding-specific extended Kafka consumer properties while customizing the listener container. + +=== Usage + +To use the `KafkaListenerContainerCustomizer`, create a bean that implements this interface in your configuration: + +[source,java] +---- +@Bean +public KafkaListenerContainerCustomizer> kafkaCustomizer() { + return (container, destinationName, group, properties) -> { + // Customize the Kafka container here + }; +} +---- + +The `KafkaListenerContainerCustomizer` interface adds the following method: + +[source,java] +---- +default void configureKafkaListenerContainer( + C container, + String destinationName, + String group, + ExtendedConsumerProperties extendedConsumerProperties) { + configure(container, destinationName, group); +} +---- + +This method extends the base `configure` method with an additional parameter: + +* `extendedConsumerProperties`: Extended consumer properties, including Kafka-specific properties. + +== ListenerContainerWithDlqAndRetryCustomizer + +The `ListenerContainerWithDlqAndRetryCustomizer` interface provides additional customization options for scenarios involving Dead Letter Queues (DLQ) and retry mechanisms. + +=== Purpose + +Use this customizer when you need to fine-tune DLQ behavior or implement custom retry logic for your Kafka consumers. + +=== Usage + +To use the `ListenerContainerWithDlqAndRetryCustomizer`, create a bean that implements this interface in your configuration: + +[source,java] +---- +@Bean +public ListenerContainerWithDlqAndRetryCustomizer dlqCustomizer() { + return (container, destinationName, group, dlqDestinationResolver, backOff, properties) -> { + // Access the container here with access to the extended consumer binding properties. + }; +} +---- + +The `ListenerContainerWithDlqAndRetryCustomizer` interface defines the following method: + +[source,java] +---- +void configure( + AbstractMessageListenerContainer container, + String destinationName, + String group, + BiFunction, Exception, TopicPartition> dlqDestinationResolver, + BackOff backOff, + ExtendedConsumerProperties extendedConsumerProperties +); +---- + +* `container`: The Kafka listener container to customize. +* `destinationName`: The name of the destination (topic). +* `group`: The consumer group ID. +* `dlqDestinationResolver`: A function to resolve the DLQ destination for a failed record. +* `backOff`: The backoff policy for retries. +* `extendedConsumerProperties`: Extended consumer properties, including Kafka-specific properties. + +== Summary + +* `ListenerContainerWithDlqAndRetryCustomizer` is used if DLQ is enabled. +* `KafkaListenerContainerCustomizer` is used for Kafka-specific customization without DLQ. +* The base `ListenerContainerCustomizer` is used for generic customization. + +This hierarchical approach allows for flexible and specific customization of your Kafka listener containers in Spring Cloud Stream applications.