Add Apache Camel support (#3887)

* Add Apache Camel support

* Implement `CamelMessageHandler` to perform send and send-n-reply
operations to the Apache Camel routes
* Fix `AbstractMessageProducingHandler` to handle async errors even if reply is
not expected

* * Narrow Camel dependency to `api`

build.gradle

@@ -54,6 +54,7 @@ ext {
     assertkVersion = '0.25'
     avroVersion = '1.11.0'
     awaitilityVersion = '4.2.0'
+    camelVersion = '3.18.2'
     commonsDbcp2Version = '2.9.0'
     commonsIoVersion = '2.11.0'
     commonsNetVersion = '3.8.0'
@@ -163,6 +164,7 @@ allprojects {
             mavenBom "org.mockito:mockito-bom:$mockitoVersion"
             mavenBom "io.micrometer:micrometer-bom:$micrometerVersion"
             mavenBom "io.micrometer:micrometer-tracing-bom:$micrometerTracingVersion"
+            mavenBom "org.apache.camel:camel-bom:$camelVersion"
         }
 
     }
@@ -477,6 +479,18 @@ project('spring-integration-amqp') {
     }
 }
 
+project('spring-integration-camel') {
+    description = 'Spring Integration support for Apache Camel'
+
+    dependencies {
+        api project(':spring-integration-core')
+        api 'org.apache.camel:camel-api'
+
+        testImplementation 'org.apache.camel:camel-test-junit5'
+    }
+}
+
+
 project('spring-integration-core') {
     description = 'Spring Integration Core'
 

spring-integration-camel/src/main/java/org/springframework/integration/camel/outbound/CamelMessageHandler.java

@@ -0,0 +1,202 @@
+/*
+ * Copyright 2022 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.camel.outbound;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.ProducerTemplate;
+
+import org.springframework.expression.Expression;
+import org.springframework.expression.common.LiteralExpression;
+import org.springframework.expression.spel.support.StandardEvaluationContext;
+import org.springframework.integration.camel.support.CamelHeaderMapper;
+import org.springframework.integration.expression.ExpressionUtils;
+import org.springframework.integration.expression.ValueExpression;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.mapping.HeaderMapper;
+import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * A {@link org.springframework.messaging.MessageHandler} for calling Apache Camel route
+ * and produce (optionally) a reply.
+ * <p>
+ * In the async mode, the {@link ProducerTemplate#asyncSend(Endpoint, Exchange)} is used.
+ * <p>
+ * The request-reply behavior can be controlled via {@link ExchangePattern} configuration
+ * or per message. By default, this handler works in an {@link ExchangePattern#InOnly} mode.
+ * <p>
+ * A default "mapping all headers" between Spring Integration and Apache Camel messages behavior
+ * can be customized via {@link  #setHeaderMapper(HeaderMapper)} option.
+ * <p>
+ * The target Apache Camel endpoint to call can be determined by the {@link #endpointUriExpression}.
+ * By default, a {@link ProducerTemplate#getDefaultEndpoint()} is used.
+ *
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ *
+ * @see CamelHeaderMapper
+ */
+public class CamelMessageHandler extends AbstractReplyProducingMessageHandler {
+
+	private final ProducerTemplate producerTemplate;
+
+	private Expression exchangePatternExpression = new ValueExpression<>(ExchangePattern.InOnly);
+
+	@Nullable
+	private Expression endpointUriExpression;
+
+	private HeaderMapper<org.apache.camel.Message> headerMapper = new CamelHeaderMapper();
+
+	@Nullable
+	private Expression exchangePropertiesExpression;
+
+	private StandardEvaluationContext evaluationContext;
+
+	public CamelMessageHandler(ProducerTemplate producerTemplate) {
+		Assert.notNull(producerTemplate, "'producerTemplate' must not be null");
+		this.producerTemplate = producerTemplate;
+	}
+
+	public void setEndpointUri(String endpointUri) {
+		Assert.hasText(endpointUri, "'endpointUri' must not be empty");
+		setEndpointUriExpression(new LiteralExpression(endpointUri));
+	}
+
+	public void setEndpointUriExpression(Expression endpointUriExpression) {
+		Assert.notNull(endpointUriExpression, "'endpointUriExpression' must not be null");
+		this.endpointUriExpression = endpointUriExpression;
+	}
+
+	public void setExchangePattern(ExchangePattern exchangePattern) {
+		Assert.notNull(exchangePattern, "'exchangePattern' must not be null");
+		setExchangePatternExpression(new ValueExpression<>(exchangePattern));
+	}
+
+	public void setExchangePatternExpression(Expression exchangePatternExpression) {
+		Assert.notNull(exchangePatternExpression, "'exchangePatternExpression' must not be null");
+		this.exchangePatternExpression = exchangePatternExpression;
+	}
+
+	/**
+	 * Set a {@link HeaderMapper} to map request message headers into Apache Camel message headers and
+	 * back if request-reply exchange pattern is used.
+	 * @param headerMapper the {@link HeaderMapper} to use.
+	 */
+	public void setHeaderMapper(HeaderMapper<org.apache.camel.Message> headerMapper) {
+		Assert.notNull(headerMapper, "'headerMapper' must not be null");
+		this.headerMapper = headerMapper;
+	}
+
+	public void setExchangeProperties(Map<String, Object> exchangeProperties) {
+		Assert.notNull(exchangeProperties, "'exchangeProperties' must not be null");
+		setExchangePropertiesExpression(new ValueExpression<>(exchangeProperties));
+	}
+
+	/**
+	 * Set a SpEL expression to evaluate {@link org.apache.camel.Exchange} properties as a {@link Map}.
+	 * @param exchangePropertiesExpression the expression for exchange properties.
+	 */
+	public void setExchangePropertiesExpression(Expression exchangePropertiesExpression) {
+		this.exchangePropertiesExpression = exchangePropertiesExpression;
+	}
+
+	@Override
+	protected final void doInit() {
+		this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
+	}
+
+	@Override
+	protected Object handleRequestMessage(Message<?> requestMessage) {
+		ExchangePattern exchangePattern =
+				this.exchangePatternExpression.getValue(this.evaluationContext, requestMessage, ExchangePattern.class);
+
+		Assert.notNull(exchangePattern, "'exchangePatternExpression' must not evaluate to null");
+
+		Endpoint endpoint = resolveEndpoint(requestMessage);
+		Exchange exchange = prepareInExchange(endpoint, exchangePattern, requestMessage);
+
+		if (isAsync()) {
+			CompletableFuture<Exchange> result = this.producerTemplate.asyncSend(endpoint, exchange);
+			return result.thenApply(resultExchange -> buildReply(exchangePattern, resultExchange));
+		}
+		else {
+			Exchange result = this.producerTemplate.send(endpoint, exchange);
+			return buildReply(exchangePattern, result);
+		}
+	}
+
+	private Endpoint resolveEndpoint(Message<?> requestMessage) {
+		String endpointUri =
+				this.endpointUriExpression != null
+						? this.endpointUriExpression.getValue(this.evaluationContext, requestMessage, String.class)
+						: null;
+
+		if (StringUtils.hasText(endpointUri)) {
+			return this.producerTemplate.getCamelContext().getEndpoint(endpointUri);
+		}
+		else {
+			return this.producerTemplate.getDefaultEndpoint();
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	private Exchange prepareInExchange(Endpoint endpoint, ExchangePattern exchangePattern, Message<?> requestMessage) {
+		Exchange exchange = endpoint.createExchange(exchangePattern);
+
+		Map<String, Object> exchangeProperties =
+				this.exchangePropertiesExpression != null
+						? this.exchangePropertiesExpression.getValue(this.evaluationContext, requestMessage, Map.class)
+						: null;
+
+		if (exchangeProperties != null) {
+			for (Map.Entry<String, Object> property : exchangeProperties.entrySet()) {
+				exchange.setProperty(property.getKey(), property.getValue());
+			}
+		}
+		org.apache.camel.Message in = exchange.getIn();
+		this.headerMapper.fromHeaders(requestMessage.getHeaders(), in);
+		in.setBody(requestMessage.getPayload());
+		return exchange;
+	}
+
+	@Nullable
+	private AbstractIntegrationMessageBuilder<?> buildReply(ExchangePattern exchangePattern, Exchange result) {
+		if (result.isFailed()) {
+			throw CamelExecutionException.wrapCamelExecutionException(result, result.getException());
+		}
+		if (exchangePattern.isOutCapable()) {
+			org.apache.camel.Message out = result.getMessage();
+			return getMessageBuilderFactory()
+					.withPayload(out.getBody())
+					.copyHeaders(this.headerMapper.toHeaders(out));
+		}
+		else {
+			return null;
+		}
+	}
+
+}

spring-integration-camel/src/main/java/org/springframework/integration/camel/outbound/package-info.java

@@ -0,0 +1,7 @@
+/**
+ * Provides classes for Apache Camel outbound channel adapters.
+ */
+
+@org.springframework.lang.NonNullApi
+@org.springframework.lang.NonNullFields
+package org.springframework.integration.camel.outbound;

spring-integration-camel/src/main/java/org/springframework/integration/camel/support/CamelHeaderMapper.java

@@ -0,0 +1,115 @@
+/*
+ * Copyright 2022 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.camel.support;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.camel.Message;
+
+import org.springframework.core.log.LogAccessor;
+import org.springframework.core.log.LogMessage;
+import org.springframework.integration.mapping.HeaderMapper;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.util.Assert;
+import org.springframework.util.PatternMatchUtils;
+
+/**
+ * A {@link HeaderMapper} for mapping headers from Spring Integration message
+ * to Apache Camel message and back.
+ *
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public class CamelHeaderMapper implements HeaderMapper<org.apache.camel.Message> {
+
+
+	private static final LogAccessor LOGGER = new LogAccessor(CamelHeaderMapper.class);
+
+	private String[] inboundHeaderNames = { "*" };
+
+	private String[] outboundHeaderNames = { "*" };
+
+	/**
+	 * Provide a list of patterns to map Apache Camel message headers into Spring Integration message.
+	 * By default, it maps all.
+	 * @param inboundHeaderNames the Apache Camel message headers patterns to map.
+	 */
+	public void setInboundHeaderNames(String... inboundHeaderNames) {
+		Assert.notNull(inboundHeaderNames, "'inboundHeaderNames' must not be null");
+		String[] copy = Arrays.copyOf(inboundHeaderNames, inboundHeaderNames.length);
+		Arrays.sort(copy);
+		this.inboundHeaderNames = copy;
+	}
+
+	/**
+	 * Provide a list of patterns to map Spring Integration message headers into an Apache Camel message.
+	 * By default, it maps all.
+	 * @param outboundHeaderNames the header patterns to map.
+	 */
+	public void setOutboundHeaderNames(String... outboundHeaderNames) {
+		Assert.notNull(outboundHeaderNames, "'outboundHeaderNames' must not be null");
+		String[] copy = Arrays.copyOf(outboundHeaderNames, outboundHeaderNames.length);
+		Arrays.sort(copy);
+		this.outboundHeaderNames = copy;
+	}
+
+	@Override
+	public void fromHeaders(MessageHeaders headers, Message target) {
+		for (Map.Entry<String, Object> entry : headers.entrySet()) {
+			String name = entry.getKey();
+			if (shouldMapHeader(name, this.outboundHeaderNames)) {
+				Object value = entry.getValue();
+				if (value != null) {
+					target.setHeader(name, value);
+				}
+			}
+		}
+	}
+
+	@Override
+	public Map<String, Object> toHeaders(Message source) {
+		Map<String, Object> headers = new HashMap<>();
+		for (Map.Entry<String, Object> entry : source.getHeaders().entrySet()) {
+			String name = entry.getKey();
+			if (shouldMapHeader(name, this.inboundHeaderNames)) {
+				Object value = entry.getValue();
+				if (value != null) {
+					headers.put(name, value);
+				}
+			}
+		}
+		return headers;
+	}
+
+	private static boolean shouldMapHeader(String headerName, String[] patterns) {
+		if (patterns.length > 0) {
+			for (String pattern : patterns) {
+				if (PatternMatchUtils.simpleMatch(pattern, headerName)) {
+					LOGGER.debug(LogMessage.format("headerName=[{0}] WILL be mapped, matched pattern={1}",
+							headerName, pattern));
+					return true;
+				}
+			}
+		}
+		LOGGER.debug(LogMessage.format("headerName=[{0}] WILL NOT be mapped", headerName));
+		return false;
+	}
+
+}

spring-integration-camel/src/main/java/org/springframework/integration/camel/support/package-info.java

@@ -0,0 +1,7 @@
+/**
+ * Provides supporting classes for Apache Camel channel adapters.
+ */
+
+@org.springframework.lang.NonNullApi
+@org.springframework.lang.NonNullFields
+package org.springframework.integration.camel.support;

spring-integration-camel/src/test/java/org/springframework/integration/camel/outbound/CamelMessageHandlerTests.java

@@ -0,0 +1,189 @@
+/*
+ * Copyright 2022 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.camel.outbound;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.assertj.core.api.InstanceOfAssertFactories;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.integration.camel.support.CamelHeaderMapper;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.expression.FunctionExpression;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.support.GenericMessage;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 6.0
+ */
+public class CamelMessageHandlerTests extends CamelTestSupport {
+
+	@Test
+	void inOnlyPatternSyncMessageHandler() throws InterruptedException {
+		Message<String> messageUnderTest = new GenericMessage<>("Hello Camel!");
+		Message<String> messageUnderTest2 = new GenericMessage<>("Hello Camel again!");
+
+		MockEndpoint mockEndpoint = getMockEndpoint("mock:result");
+		mockEndpoint.message(0).body().isEqualTo(messageUnderTest.getPayload());
+		mockEndpoint.message(0).header(MessageHeaders.ID).isEqualTo(messageUnderTest.getHeaders().getId());
+		mockEndpoint.message(1).body().isEqualTo(messageUnderTest2.getPayload());
+		mockEndpoint.message(1).header(MessageHeaders.ID).isEqualTo(messageUnderTest2.getHeaders().getId());
+
+		CamelMessageHandler camelMessageHandler = new CamelMessageHandler(template());
+		camelMessageHandler.setEndpointUri("direct:simple");
+		camelMessageHandler.setBeanFactory(mock(BeanFactory.class));
+		camelMessageHandler.afterPropertiesSet();
+
+		camelMessageHandler.handleMessage(messageUnderTest);
+		camelMessageHandler.handleMessage(messageUnderTest2);
+
+		assertMockEndpointsSatisfied();
+	}
+
+	@Test
+	void inOutPatternSyncMessageHandlerWithNoRequestHeadersButReplyHeaders() throws InterruptedException {
+		SpelExpressionParser spelExpressionParser = new SpelExpressionParser();
+		QueueChannel replyChannel = new QueueChannel();
+		Message<String> messageUnderTest =
+				MessageBuilder.withPayload("test data")
+						.setHeader("exchangePattern", "InOptionalOut")
+						.setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel)
+						.build();
+
+		MockEndpoint mockEndpoint = getMockEndpoint("mock:result");
+		mockEndpoint.expectedHeaderReceived(MessageHeaders.ID, null);
+		mockEndpoint.expectedHeaderReceived(MessageHeaders.TIMESTAMP, null);
+		mockEndpoint.whenAnyExchangeReceived(exchange -> {
+			org.apache.camel.Message out = exchange.getMessage();
+			out.setBody("Reply for: " + exchange.getIn().getBody());
+			out.setHeader("testHeader", "testHeaderValue");
+			out.setHeader("notMappedHeader", "someValue");
+		});
+
+		CamelHeaderMapper headerMapper = new CamelHeaderMapper();
+		headerMapper.setOutboundHeaderNames("");
+		headerMapper.setInboundHeaderNames("testHeader");
+
+		CamelMessageHandler camelMessageHandler = new CamelMessageHandler(template());
+		camelMessageHandler.setEndpointUriExpression(new FunctionExpression<>(m -> "direct:simple"));
+		camelMessageHandler.setExchangePatternExpression(spelExpressionParser.parseExpression("headers.exchangePattern"));
+		camelMessageHandler.setHeaderMapper(headerMapper);
+		camelMessageHandler.setBeanFactory(mock(BeanFactory.class));
+		camelMessageHandler.afterPropertiesSet();
+
+		camelMessageHandler.handleMessage(messageUnderTest);
+
+		Message<?> receive = replyChannel.receive(10_000);
+
+		assertMockEndpointsSatisfied();
+
+		assertThat(receive).isNotNull();
+		assertThat(receive.getPayload()).isEqualTo("Reply for: test data");
+		assertThat(receive.getHeaders())
+				.containsEntry("testHeader", "testHeaderValue")
+				.doesNotContainKey("notMappedHeader");
+	}
+
+	@Test
+	void inOnlyPatternAsyncMessageHandlerWithException() throws InterruptedException {
+		QueueChannel errorChannel = new QueueChannel();
+		Message<String> messageUnderTest =
+				MessageBuilder.withPayload("test data")
+						.setHeader(MessageHeaders.ERROR_CHANNEL, errorChannel)
+						.build();
+
+		getMockEndpoint("mock:result")
+				.whenAnyExchangeReceived(exchange -> {
+					throw new RuntimeException("intentional");
+				});
+
+		CamelMessageHandler camelMessageHandler = new CamelMessageHandler(template());
+		camelMessageHandler.setEndpointUri("direct:simple");
+		camelMessageHandler.setBeanFactory(mock(BeanFactory.class));
+		camelMessageHandler.setAsync(true);
+		camelMessageHandler.afterPropertiesSet();
+
+		camelMessageHandler.handleMessage(messageUnderTest);
+		Message<?> receive = errorChannel.receive(10_000);
+
+		assertMockEndpointsSatisfied();
+
+		assertThat(receive).isNotNull();
+		assertThat(receive.getPayload())
+				.asInstanceOf(InstanceOfAssertFactories.throwable(MessageHandlingException.class))
+				.hasCauseInstanceOf(CamelExecutionException.class)
+				.hasRootCauseExactlyInstanceOf(RuntimeException.class)
+				.hasStackTraceContaining("intentional");
+	}
+
+	@Test
+	void inOutPatternAsyncMessageHandler() throws InterruptedException {
+		QueueChannel replyChannel = new QueueChannel();
+		Message<String> messageUnderTest =
+				MessageBuilder.withPayload("test async data")
+						.setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel)
+						.build();
+
+		MockEndpoint mockEndpoint = getMockEndpoint("mock:result");
+		mockEndpoint.whenAnyExchangeReceived(exchange ->
+				exchange.getMessage().setBody("Async reply for: " + exchange.getIn().getBody()));
+
+		ProducerTemplate producerTemplate = template();
+		producerTemplate.setDefaultEndpointUri("direct:simple");
+		CamelMessageHandler camelMessageHandler = new CamelMessageHandler(producerTemplate);
+		camelMessageHandler.setExchangePattern(ExchangePattern.InOut);
+		camelMessageHandler.setBeanFactory(mock(BeanFactory.class));
+		camelMessageHandler.setAsync(true);
+		camelMessageHandler.afterPropertiesSet();
+
+		camelMessageHandler.handleMessage(messageUnderTest);
+
+		Message<?> receive = replyChannel.receive(10_000);
+
+		assertMockEndpointsSatisfied();
+
+		assertThat(receive).isNotNull();
+		assertThat(receive.getPayload()).isEqualTo("Async reply for: test async data");
+	}
+
+	@Override
+	protected RoutesBuilder createRouteBuilder() {
+		return new RouteBuilder() {
+
+			@Override
+			public void configure() {
+				from("direct:simple").to("mock:result");
+			}
+		};
+	}
+
+}

spring-integration-camel/src/test/resources/log4j2-test.xml

@@ -0,0 +1,15 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration status="WARN">
+	<Appenders>
+		<Console name="STDOUT" target="SYSTEM_OUT">
+			<PatternLayout pattern="%d %p [%t] [%c] - %m%n" />
+		</Console>
+	</Appenders>
+	<Loggers>
+		<Logger name="org.springframework.integration" level="warn"/>
+		<Logger name="org.springframework.integration.camel" level="info"/>
+		<Root level="warn">
+			<AppenderRef ref="STDOUT" />
+		</Root>
+	</Loggers>
+</Configuration>

spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java

@@ -24,6 +24,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiConsumer;
 
@@ -535,7 +536,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 
 		@Override
 		public void accept(Object result, Throwable exception) {
-			if (exception == null) {
+			if (result != null) {
 				Message<?> replyMessage = null;
 				try {
 					replyMessage = createOutputMessage(result, this.requestMessage.getHeaders());
@@ -549,12 +550,12 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 							exceptionToLogAndSend = new MessagingException(replyMessage, exceptionToLogAndSend);
 						}
 					}
-					logger.error(exceptionToLogAndSend, () -> "Failed to send async reply: " + result.toString());
+					logger.error(exceptionToLogAndSend, () -> "Failed to send async reply: " + result);
 					onFailure(exceptionToLogAndSend);
 				}
 			}
-			else {
-				onFailure(exception);
+			else if (exception != null) {
+				onFailure(exception instanceof CompletionException ? exception.getCause() : exception);
 			}
 		}