SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #3015 from sobychacko/gh-2985

Browse Source 
GH-2985: Add Kafka Listener Container Customizer interfaces and docum…
This commit is contained in:
Oleg Zhurakousky
2024-10-07 18:43:26 +02:00
committed by GitHub
parent cbfd3aa995 f8d6caa429
commit eea905f692
6 changed files with 372 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaListenerContainerCustomizer.java Normal file
View File

@@ -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);
}
}

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

@@ -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)

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

@@ -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

162
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaListenerContainerCustomizerTests.java Normal file
View File

@@ -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) {
}
}
}
}