Add setter for Container Customizer
Add setter for Container Customizer on AbstractMessageChannelBinder. Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2345
This commit is contained in:
committed by
Soby Chacko
parent
ad89acf62d
commit
8682417af4
@@ -906,6 +906,28 @@ ListenerContainerWithDlqAndRetryCustomizer cust(KafkaTemplate<?, ?> template) {
|
||||
|
||||
Now, only a single retry delay needs to be greater than the consumer's `max.poll.interval.ms` property.
|
||||
|
||||
When working with several binders, the 'ListenerContainerWithDlqAndRetryCustomizer' bean gets overridden by the 'DefaultBinderFactory'. For the bean
|
||||
to apply, you need to use a 'BinderCustomizer' to set the container customizer (See <<binder-customizer>>):
|
||||
====
|
||||
[source, java]
|
||||
----
|
||||
@Bean
|
||||
public BinderCustomizer binderCustomizer(ListenerContainerWithDlqAndRetryCustomizer containerCustomizer) {
|
||||
return (binder, binderName) -> {
|
||||
if (binder instanceof KafkaMessageChannelBinder) {
|
||||
((KafkaMessageChannelBinder) binder).setContainerCustomizer(containerCustomizer);
|
||||
}
|
||||
else if (binder instanceof KStreamBinder) {
|
||||
...
|
||||
}
|
||||
else if (binder instanceof RabbitMessageChannelBinder) {
|
||||
...
|
||||
}
|
||||
};
|
||||
}
|
||||
----
|
||||
====
|
||||
|
||||
[[consumer-producer-config-customizer]]
|
||||
=== Customizing Consumer and Producer configuration
|
||||
|
||||
|
||||
@@ -113,7 +113,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
|
||||
*/
|
||||
private final String[] headersToEmbed;
|
||||
|
||||
private final ListenerContainerCustomizer<?> containerCustomizer;
|
||||
private ListenerContainerCustomizer<?> containerCustomizer;
|
||||
|
||||
private final MessageSourceCustomizer<?> sourceCustomizer;
|
||||
|
||||
@@ -197,6 +197,20 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
|
||||
: (ConsumerEndpointCustomizer<MessageProducer>) endpointCustomizer;
|
||||
}
|
||||
|
||||
/**
|
||||
* Configure an optional {@link ListenerContainerCustomizer} for further
|
||||
* configuration of the listener container instance created by the binder.
|
||||
* @param containerCustomizer the {@link ListenerContainerCustomizer} to use.
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public void setContainerCustomizer(@Nullable ListenerContainerCustomizer<?> containerCustomizer) {
|
||||
|
||||
this.containerCustomizer =
|
||||
containerCustomizer == null
|
||||
? (container, destinationName, group) -> { }
|
||||
: containerCustomizer;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
protected <L> ListenerContainerCustomizer<L> getContainerCustomizer() {
|
||||
return (ListenerContainerCustomizer<L>) this.containerCustomizer;
|
||||
|
||||
Reference in New Issue
Block a user