Partition selector strategy does not use the configured partition count for multiple outbounds

[angelbeshirov]

Framework Version:
spring-boot: 3.0.2
spring-cloud: 2022.0.1

Issue:
We have a reactive function, which consumes messages from one topic and sends them to N topics based on some conditions. Additionally, we also have a custom key extractor and partition selector configured for each of the output bindings. However, the number of partitions of the outbound topics is different which does not seem to be reflected in the custom partition selector.

I did some digging and it seems only the binding properties of the first outbound are being used when creating the PartitionAwareFunctionWrapper, which are then passed to the PartitionHandler invoking the overriden selectPartition(Object key, int partitionCount) method with incorrect partition count.

Sending the message to the incorrect partition results in TimeoutException.

Code snippet from org.springframework.cloud.stream.function.FunctionConfiguration

if (!CollectionUtils.isEmpty(outputBindingNames)) {
	BindingProperties bindingProperties = this.serviceProperties.getBindings().get(outputBindingNames.iterator().next());
	ProducerProperties producerProperties = bindingProperties == null ? null : bindingProperties.getProducer();
	if (producerProperties != null) {
		function.setSkipOutputConversion(producerProperties.isUseNativeEncoding());
	}
	functionToInvoke = new PartitionAwareFunctionWrapper(function, this.applicationContext, producerProperties);
}

Expected behavior:
I would expect the framework to honor the configured partition count for each output binding.

[sobychacko]

How do you send them to N topics? Could you provide us with a small sample application where we can see the issue?

[angelbeshirov]

Hi @sobychacko

I have created a sample application showing the issue: https://github.com/angelbeshirov/spring-cloud-partition-selector-issue

The reactive function reads random numbers from one topic test-input and sends the even numbers to output-even and the odd ones to output-odd. The output-even topic has 10 partitions and output-odd has 5 partitions.

Publishing a message to output-odd fails with:

Exception thrown when sending a message with key='null' and payload='byte[30]' to topic output-odd and partition 9:
org.apache.kafka.common.errors.TimeoutException: Topic output-odd not present in metadata after 60000 ms.

[sobychacko]

@angelbeshirov The issue is now resolved via the commit mentioned ^^. You can try the fix on the latest snapshots.

[angelbeshirov]

Thanks, @sobychacko, I can confirm the issue is resolved in 4.1.0-SNAPSHOT.

[sobychacko]

Thanks @angelbeshirov for confirming.