GH-2062 Add logic to log an error for reactive functions

Resolves #2062

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

@@ -472,6 +472,9 @@ public class FunctionConfiguration {
 								}
 								outputChannel.send((Message) message);
 							}
+						}).doOnError(e -> {
+							logger.error("Failure was detected during execution of the reactive function '" +  functionDefinition + "'");
+							((Throwable) e).printStackTrace();
 						});
 					}
 					if (!function.isConsumer()) {

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java

@@ -909,6 +909,20 @@ public class ImplicitFunctionBindingTests {
 	}
 
 
+	@Test
+	public void testGh2062() {
+		System.clearProperty("spring.cloud.function.definition");
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(ReactiveFunctionConfiguration.class))
+						.web(WebApplicationType.NONE).run("--spring.jmx.enabled=false",
+								"--spring.cloud.function.definition=echo")) {
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			String jsonPerson = "error";
+
+			inputDestination.send(MessageBuilder.withPayload(jsonPerson.getBytes()).build());
+			// there is really nothing to assert other then check the logs for stack trace and error message
+		}
+	}
 
 	@SuppressWarnings("rawtypes")
 	@Test
@@ -1181,6 +1195,9 @@ public class ImplicitFunctionBindingTests {
 		public Function<Flux<String>, Flux<String>> echo() {
 			return flux -> flux.map(value -> {
 				System.out.println("echo value reqctive " + value);
+				if (value.equals("error")) {
+					throw new RuntimeException("intentional");
+				}
 				return value;
 			});
 		}