GH-2985: Add Kafka Listener Container Customizer interfaces and documentation
Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2985 This commit introduces new customization options for Kafka listener containers in Spring Cloud Stream, along with comprehensive documentation: - Add KafkaListenerContainerCustomizer interface for Kafka-specific customization with access to extended consumer properties - Extend ListenerContainerWithDlqAndRetryCustomizer to include access to extended consumer properties - Update KafkaMessageChannelBinder to support the new customizer interfaces - Implement KafkaListenerContainerCustomizerTests for integration testing - Add detailed AsciiDoc reference documentation explaining the purpose, usage, and hierarchy of these customizer interfaces: * ListenerContainerCustomizer (existing) * KafkaListenerContainerCustomizer (new) * ListenerContainerWithDlqAndRetryCustomizer (extended) - Update navigation to include the new documentation These changes enhance the flexibility and configurability of Kafka consumer endpoints in Spring Cloud Stream applications, allowing users to fine-tune their listener containers based on specific requirements and scenarios, with improved access to Kafka-specific properties.
This commit is contained in:
Soby Chacko
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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<Object>) customizer)
|
||||
|
||||
@@ -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<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 @@ public interface ListenerContainerWithDlqAndRetryCustomizer
|
||||
@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
|
||||
|
||||
@@ -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) {
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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[]
|
||||
|
||||
@@ -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<AbstractMessageListenerContainer<?, ?>> 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<AbstractMessageListenerContainer<?, ?>> 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<KafkaConsumerProperties> 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<ConsumerRecord<?, ?>, Exception, TopicPartition> dlqDestinationResolver,
|
||||
BackOff backOff,
|
||||
ExtendedConsumerProperties<KafkaConsumerProperties> 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.
|
||||
Reference in New Issue
Block a user