From 7840decc867ca5d8dd79d5797136c2875425e85c Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 16 Nov 2021 14:43:43 +0100 Subject: [PATCH] Changes related to GH-2245 from core --- .../cloud/stream/binder/kafka/KafkaMessageChannelBinder.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 5334e2eff..b3211ed26 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -76,6 +76,7 @@ import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerPro import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; import org.springframework.cloud.stream.binder.kafka.utils.DlqDestinationResolver; import org.springframework.cloud.stream.binder.kafka.utils.DlqPartitionFunction; +import org.springframework.cloud.stream.binding.DefaultPartitioningInterceptor; import org.springframework.cloud.stream.binding.MessageConverterConfigurer.PartitioningInterceptor; import org.springframework.cloud.stream.config.ListenerContainerCustomizer; import org.springframework.cloud.stream.config.MessageSourceCustomizer; @@ -418,7 +419,7 @@ public class KafkaMessageChannelBinder extends List interceptors = ((InterceptableChannel) channel) .getInterceptors(); interceptors.forEach((interceptor) -> { - if (interceptor instanceof PartitioningInterceptor) { + if (interceptor instanceof PartitioningInterceptor || interceptor instanceof DefaultPartitioningInterceptor) { ((PartitioningInterceptor) interceptor) .setPartitionCount(partitions.size()); }