mindol1004/spring-cloud-stream-binder-kafka
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix PartitioningInterceptor CCE

Browse Source 
The newly added DefaultPartitioningInteceptor must be explicitly
checked in order to avoid a CCE.

Related to resolving https://github.com/spring-cloud/spring-cloud-stream/issues/2245

Specifically for this: https://github.com/spring-cloud/spring-cloud-stream/issues/2245#issuecomment-977663452
This commit is contained in:
Soby Chacko
2021-11-24 13:12:52 -05:00
parent e9a8b4af7e
commit e7bf404fce
1 changed files with 5 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -419,10 +419,14 @@ public class KafkaMessageChannelBinder extends
List<ChannelInterceptor> interceptors = ((InterceptableChannel) channel) List<ChannelInterceptor> interceptors = ((InterceptableChannel) channel)
.getInterceptors(); .getInterceptors();
interceptors.forEach((interceptor) -> { interceptors.forEach((interceptor) -> {
if (interceptor instanceof PartitioningInterceptor || interceptor instanceof DefaultPartitioningInterceptor) { if (interceptor instanceof PartitioningInterceptor) {
((PartitioningInterceptor) interceptor) ((PartitioningInterceptor) interceptor)
.setPartitionCount(partitions.size()); .setPartitionCount(partitions.size());
} }
else if (interceptor instanceof DefaultPartitioningInterceptor) {
((DefaultPartitioningInterceptor) interceptor)
.setPartitionCount(partitions.size());
}
}); });
} }