StreamBridge does not allow to switch between different producer configs due to internal caching

We found the following potential problem while experimenting with different producer configurations - some with partitionKeyExpression and other without.

StreamBridge uses internal function cache:
private final Map<String, FunctionInvocationWrapper> streamBridgeFunctionCache;

This function cache is often overwriten (not used), if a producer configuration "isPartitioned":

Function functionToInvoke = this.getStreamBridgeFunction(outputContentType.toString(), producerProperties);
if (producerProperties != null && producerProperties.isPartitioned())
 { functionToInvoke = new PartitionAwareFunctionWrapper(functionToInvoke, this.applicationContext, producerProperties);}

With this implementation it is not possible to use different producer configurations e.g. which are not partitioned at the same time.

The cache should be removed or the key to the cache should be the name of the producer config.

This is no urgent issue for at the moment.

Start by locating the StreamBridge implementation and inspect streamBridgeFunctionCache, getStreamBridgeFunction, and the PartitionAwareFunctionWrapper path. Reproduce the issue with simultaneous partitioned and non-partitioned producer configurations. Done means different producer configurations can be switched between and used without the internal cache causing one configuration to overwrite another.