GH-2847: Multiple outputs routing-key-expression

- When there are multiple output bindings present and one of them defines
   a routing-key-expression, there is a bug that bypasses the code that skips
   the output conversion. This results in the framework attempts a pre-mature
   type conversion causing in later downstream errors. This happens because
   MessageConverterConfigurer tries to find a corresponding function for the
   entire function definition rather than using the individual function under
   consideraion. Fixing this issue by properly keeping track of the function
   name keyed off of the channel name, since channel name is what
   MessageConverterConfigurer uses to retrieve info about the function name.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2847

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2020 the original author or authors.
+ * Copyright 2015-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -35,6 +35,7 @@ import org.springframework.cloud.stream.binder.ProducerProperties;
 import org.springframework.cloud.stream.config.BindingProperties;
 import org.springframework.cloud.stream.config.BindingServiceProperties;
 import org.springframework.cloud.stream.converter.MessageConverterUtils;
+import org.springframework.cloud.stream.function.BindableFunctionProxyFactory;
 import org.springframework.cloud.stream.function.StreamFunctionProperties;
 import org.springframework.core.env.Environment;
 import org.springframework.integration.channel.AbstractMessageChannel;
@@ -164,7 +165,9 @@ public class MessageConverterConfigurer
 			}
 			else {
 				if (environment != null && environment.containsProperty("spring.cloud.stream.rabbit.bindings." + channelName + ".producer.routing-key-expression")) {
-					this.setSkipOutputConversionIfNecessary();
+					Map<String, String> channelNameToFunctions = BindableFunctionProxyFactory.getChannelNameToFunctions();
+					String functionName = channelNameToFunctions.get(channelName);
+					this.setSkipOutputConversionIfNecessary(functionName);
 					functional = false;
 				}
 				if (!functional) {
@@ -174,10 +177,10 @@ public class MessageConverterConfigurer
 		}
 	}
 
-	private void setSkipOutputConversionIfNecessary() {
+	private void setSkipOutputConversionIfNecessary(String functionName) {
 		FunctionCatalog catalog = this.beanFactory.getBean(FunctionCatalog.class);
 		if (catalog != null) {
-			FunctionInvocationWrapper function = catalog.lookup(this.streamFunctionProperties.getDefinition());
+			FunctionInvocationWrapper function = catalog.lookup(functionName);
 			if (function != null) {
 				function.setSkipOutputConversion(true);
 			}

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2021 the original author or authors.
+ * Copyright 2019-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,6 +16,9 @@
 
 package org.springframework.cloud.stream.function;
 
+import java.util.HashMap;
+import java.util.Map;
+
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.FactoryBean;
 import org.springframework.cloud.stream.binder.PollableMessageSource;
@@ -62,6 +65,14 @@ public class BindableFunctionProxyFactory extends BindableProxyFactory implement
 
 	private GenericApplicationContext context;
 
+	/**
+	 * A convenient table containing the channel-name to function-name relationship.
+	 * This will be accessed via a utility method later on during the binding phase.
+	 * It is provided to assist with identifying the correct function that a channel
+	 * binds to when there are multiple functions present in the function definition.
+	 */
+	private static Map<String, String> channelNameToFunctions = new HashMap<>();
+
 	public BindableFunctionProxyFactory(String functionDefinition, int inputCount, int outputCount, StreamFunctionProperties functionProperties) {
 		this(functionDefinition, inputCount, outputCount, functionProperties, new SupportedBindableFeatures(), true);
 	}
@@ -166,6 +177,7 @@ public class BindableFunctionProxyFactory extends BindableProxyFactory implement
 		if (this.functionProperties.getBindings().containsKey(name)) {
 			name = this.functionProperties.getBindings().get(name);
 		}
+		updateChannelNameToFunctionName(name);
 		if (this.supportedBindableFeatures.isPollable()) {
 			PollableMessageSource pollableSource = (PollableMessageSource) getBindingTargetFactory(PollableMessageSource.class).createInput(name);
 			if (context != null && !context.containsBean(name)) {
@@ -185,10 +197,18 @@ public class BindableFunctionProxyFactory extends BindableProxyFactory implement
 		}
 	}
 
+	private void updateChannelNameToFunctionName(String name) {
+		// Update the channelName -> functionName table. Even if the application provides
+		// multiple functions in the function definition configuration (function1;function2;function3 etc.),
+		// this proxy factory only knows about a single function.
+		channelNameToFunctions.put(name, this.functionDefinition);
+	}
+
 	private void createOutput(String name) {
 		if (this.functionProperties.getBindings().containsKey(name)) {
 			name = this.functionProperties.getBindings().get(name);
 		}
+		updateChannelNameToFunctionName(name);
 		if (this.supportedBindableFeatures.isReactive()) {
 			this.outputHolders.put(name,
 				new BoundTargetHolder(getBindingTargetFactory(FluxMessageChannel.class)
@@ -209,4 +229,8 @@ public class BindableFunctionProxyFactory extends BindableProxyFactory implement
 	public boolean isFunctionExist() {
 		return functionExist;
 	}
+
+	public static Map<String, String> getChannelNameToFunctions() {
+		return channelNameToFunctions;
+	}
 }