GH-1117 Enhance support for function composition to handle null returns

Resolves #1117

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java

@@ -725,7 +725,7 @@ Object doApply(Object input) {
 
 			input = this.fluxifyInputIfNecessary(input);
 
-			Object convertedInput = this.convertInputIfNecessary(input, this.inputType);
+			Object convertedInput = input == null ? null : this.convertInputIfNecessary(input, this.inputType);
 
 			if (this.isRoutingFunction() || this.isComposed()) {
 				result = ((Function) this.target).apply(convertedInput);

spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistryTests.java

@@ -115,6 +115,15 @@ public void testEmptyPojoConversion() {
 		assertThat(result).isEqualTo("{}");
 	}
 
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Test
+	public void testCompositionWithNullReturnInBetween() {
+		FunctionCatalog catalog = this.configureCatalog(CompositionWithNullReturnInBetween.class);
+		Function function = catalog.lookup("echo1|echo2");
+		String result = (String) function.apply(MessageBuilder.withPayload(new EmptyPojo()).build());
+		assertThat(result).isEqualTo("null");
+	}
+
 	@Test
 	public void testFunctionEligibilityFiltering() {
 		System.setProperty("spring.cloud.function.ineligible-definitions", "asJsonNode");
@@ -1455,4 +1464,24 @@ public Function<String, String> echo() {
 	public static class EmptyPojo {
 
 	}
+
+	@EnableAutoConfiguration
+	@Configuration
+	public static class CompositionWithNullReturnInBetween {
+
+		@Bean
+		public Function<String, String> echo1() {
+			return v -> null;
+		}
+		@Bean
+		public Function<String, String> echo2() {
+			return v -> {
+				if (v == null) {
+					return "null";
+				}
+				return v;
+			};
+		}
+	}
+
 }