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

Changes related to GH-2245 from core

Browse Source
This commit is contained in:
Oleg Zhurakousky
2021-11-16 14:43:43 +01:00
parent d37cbc79d7
commit 7840decc86
1 changed files with 2 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<ChannelInterceptor> interceptors = ((InterceptableChannel) channel)
.getInterceptors();
interceptors.forEach((interceptor) -> {
if (interceptor instanceof PartitioningInterceptor) {
if (interceptor instanceof PartitioningInterceptor || interceptor instanceof DefaultPartitioningInterceptor) {
((PartitioningInterceptor) interceptor)
.setPartitionCount(partitions.size());
}