INT-4101: failedMessage on TX failure in pollers

JIRA: https://jira.spring.io/browse/INT-4101

Document `PassThroughTransactionSynchronizationFactory`

Copyright year and author name added

Documentation improved

Polishing

spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -33,15 +33,15 @@ import org.springframework.integration.support.channel.BeanFactoryChannelResolve
 import org.springframework.integration.transaction.ExpressionEvaluatingTransactionSynchronizationProcessor;
 import org.springframework.integration.transaction.IntegrationResourceHolder;
 import org.springframework.integration.transaction.IntegrationResourceHolderSynchronization;
+import org.springframework.integration.transaction.PassThroughTransactionSynchronizationFactory;
 import org.springframework.integration.transaction.TransactionSynchronizationFactory;
 import org.springframework.integration.util.ErrorHandlingTaskExecutor;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
-import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.MessagingException;
-import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.scheduling.Trigger;
 import org.springframework.scheduling.support.PeriodicTrigger;
+import org.springframework.transaction.interceptor.TransactionInterceptor;
 import org.springframework.transaction.support.TransactionSynchronization;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.Assert;
@@ -54,6 +54,7 @@ import org.springframework.util.ErrorHandler;
  * @author Oleg Zhurakousky
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Andreas Baer
  */
 public abstract class AbstractPollingEndpoint extends AbstractEndpoint implements BeanClassLoaderAware {
 
@@ -110,7 +111,8 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
 		this.beanClassLoader = classLoader;
 	}
 
-	public void setTransactionSynchronizationFactory(TransactionSynchronizationFactory transactionSynchronizationFactory) {
+	public void setTransactionSynchronizationFactory(TransactionSynchronizationFactory
+			transactionSynchronizationFactory) {
 		this.transactionSynchronizationFactory = transactionSynchronizationFactory;
 	}
 
@@ -168,6 +170,11 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
 					this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, this.errorHandler);
 				}
 			}
+			if (this.transactionSynchronizationFactory == null && this.adviceChain != null) {
+				if (this.adviceChain.stream().anyMatch(TransactionInterceptor.class::isInstance)) {
+					this.transactionSynchronizationFactory = new PassThroughTransactionSynchronizationFactory();
+				}
+			}
 			this.initialized = true;
 		}
 	}
@@ -300,19 +307,19 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
 	}
 
 	private IntegrationResourceHolder bindResourceHolderIfNecessary(String key, Object resource) {
-
-		if (this.transactionSynchronizationFactory != null && resource != null) {
-			if (TransactionSynchronizationManager.isActualTransactionActive()) {
-				TransactionSynchronization synchronization = this.transactionSynchronizationFactory.create(resource);
-				TransactionSynchronizationManager.registerSynchronization(synchronization);
-				if (synchronization instanceof IntegrationResourceHolderSynchronization) {
-					IntegrationResourceHolder holder =
-							((IntegrationResourceHolderSynchronization) synchronization).getResourceHolder();
-					if (key != null) {
-						holder.addAttribute(key, resource);
-					}
-					return holder;
+		if (this.transactionSynchronizationFactory != null && resource != null &&
+				TransactionSynchronizationManager.isActualTransactionActive()) {
+			TransactionSynchronization synchronization = this.transactionSynchronizationFactory.create(resource);
+			TransactionSynchronizationManager.registerSynchronization(synchronization);
+			if (synchronization instanceof IntegrationResourceHolderSynchronization) {
+				IntegrationResourceHolderSynchronization integrationSynchronization =
+						((IntegrationResourceHolderSynchronization) synchronization);
+				integrationSynchronization.setShouldUnbindAtCompletion(false);
+				IntegrationResourceHolder resourceHolder = integrationSynchronization.getResourceHolder();
+				if (key != null) {
+					resourceHolder.addAttribute(key, resource);
 				}
+				return resourceHolder;
 			}
 		}
 		return null;
@@ -343,11 +350,26 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
 						count++;
 					}
 					catch (Exception e) {
-						if (e instanceof RuntimeException) {
-							throw (RuntimeException) e;
+						if (e instanceof MessagingException) {
+							throw (MessagingException) e;
 						}
 						else {
-							throw new MessageHandlingException(new ErrorMessage(e), e);
+							Message<?> failedMessage = null;
+							if (AbstractPollingEndpoint.this.transactionSynchronizationFactory != null) {
+								Object resource = TransactionSynchronizationManager.getResource(getResourceToBind());
+								if (resource instanceof IntegrationResourceHolder) {
+									failedMessage = ((IntegrationResourceHolder) resource).getMessage();
+								}
+							}
+							throw new MessagingException(failedMessage, e);
+						}
+					}
+					finally {
+						if (AbstractPollingEndpoint.this.transactionSynchronizationFactory != null) {
+							Object resource = getResourceToBind();
+							if (TransactionSynchronizationManager.hasResource(resource)) {
+								TransactionSynchronizationManager.unbindResource(resource);
+							}
 						}
 					}
 				}

spring-integration-core/src/main/java/org/springframework/integration/transaction/IntegrationResourceHolderSynchronization.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2016 the original author or authors.
+ * Copyright 2014-2017 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.
@@ -22,13 +22,17 @@ import org.springframework.transaction.support.ResourceHolderSynchronization;
  * The base {@link ResourceHolderSynchronization} for {@link IntegrationResourceHolder}.
  *
  * @author Artem Bilan
+ * @author Andreas Baer
+ *
  * @since 4.0
  */
-public abstract class IntegrationResourceHolderSynchronization
+public class IntegrationResourceHolderSynchronization
 		extends ResourceHolderSynchronization<IntegrationResourceHolder, Object> {
 
 	protected final IntegrationResourceHolder resourceHolder;
 
+	private boolean shouldUnbindAtCompletion = true;
+
 	public IntegrationResourceHolderSynchronization(IntegrationResourceHolder resourceHolder,
 			Object resourceKey) {
 		super(resourceHolder, resourceKey);
@@ -39,4 +43,20 @@ public abstract class IntegrationResourceHolderSynchronization
 		return this.resourceHolder;
 	}
 
+	/**
+	 * Specify if the {@link #resourceHolder} should be unbound from the Thread Local store
+	 * at transaction completion or not. Default {@code true}.
+	 * @param shouldUnbindAtCompletion unbind or not {@link #resourceHolder}
+	 * at transaction completion
+	 * @since 5.0
+	 */
+	public void setShouldUnbindAtCompletion(boolean shouldUnbindAtCompletion) {
+		this.shouldUnbindAtCompletion = shouldUnbindAtCompletion;
+	}
+
+	@Override
+	protected boolean shouldUnbindAtCompletion() {
+		return this.shouldUnbindAtCompletion;
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/transaction/PassThroughTransactionSynchronizationFactory.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2017 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
+ *
+ *      http://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.transaction;
+
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+import org.springframework.util.Assert;
+
+/**
+ * A simple {@link TransactionSynchronizationFactory} implementation which produces
+ * an {@link IntegrationResourceHolderSynchronization} and registers
+ * an {@link IntegrationResourceHolder} under the provided {@code key} with
+ * the current transaction scope.
+ *
+ * @author Andreas Baer
+ *
+ * @since 5.0
+ *
+ * @see TransactionSynchronizationManager#bindResource(Object, Object)
+ */
+public class PassThroughTransactionSynchronizationFactory implements TransactionSynchronizationFactory {
+
+
+	@Override
+	public TransactionSynchronization create(Object key) {
+		Assert.notNull(key, "'key' must not be null");
+		IntegrationResourceHolderSynchronization synchronization =
+				new IntegrationResourceHolderSynchronization(new IntegrationResourceHolder(), key);
+		TransactionSynchronizationManager.bindResource(key, synchronization.getResourceHolder());
+		return synchronization;
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java

@@ -50,6 +50,7 @@ import org.springframework.scheduling.support.PeriodicTrigger;
 /**
  * @author Mark Fisher
  * @author Artem Bilan
+ * @author Andreas Baer
  */
 public class ApplicationContextMessageBusTests {
 
@@ -205,7 +206,7 @@ public class ApplicationContextMessageBusTests {
 		assertNotNull("message should not be null", message);
 		assertTrue(message instanceof ErrorMessage);
 		Throwable exception = ((ErrorMessage) message).getPayload();
-		assertEquals("intentional test failure", exception.getMessage());
+		assertEquals("intentional test failure", exception.getCause().getMessage());
 	}
 
 	@Test

spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingTransactionTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -38,16 +38,21 @@ import org.springframework.integration.test.util.TestUtils;
 import org.springframework.integration.util.TestTransactionManager;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.transaction.IllegalTransactionStateException;
+import org.springframework.transaction.TransactionException;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.interceptor.TransactionInterceptor;
+import org.springframework.transaction.support.DefaultTransactionStatus;
 
 /**
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Andreas Baer
  */
 public class PollingTransactionTests {
 
@@ -84,7 +89,8 @@ public class PollingTransactionTests {
 		Advisor[] advisors = ((Advised) pollingTask).getAdvisors();
 		assertEquals(3, advisors.length);
 
-		assertTrue("First advisor is not TX", ((DefaultPointcutAdvisor) advisors[0]).getAdvice() instanceof TransactionInterceptor);
+		assertTrue("First advisor is not TX", ((DefaultPointcutAdvisor) advisors[0]).getAdvice() instanceof
+				TransactionInterceptor);
 		TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager");
 		MessageChannel input = (MessageChannel) context.getBean("goodInputWithAdvice");
 		PollableChannel output = (PollableChannel) context.getBean("output");
@@ -196,17 +202,66 @@ public class PollingTransactionTests {
 		Message<?> errorMessage = errorChannel.receive(3000);
 		assertNotNull(errorMessage);
 		Object payload = errorMessage.getPayload();
-		assertEquals(IllegalTransactionStateException.class, payload.getClass());
+		assertEquals(MessagingException.class, payload.getClass());
+		MessagingException messagingException = (MessagingException) payload;
+		assertEquals(IllegalTransactionStateException.class, messagingException.getCause().getClass());
 		assertNull(output.receive(0));
 		assertEquals(0, txManager.getCommitCount());
 		context.close();
 	}
 
+	@Test
+	public void commitFailureAndHandlerFailureTest() throws Throwable {
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+				"transactionFailureTests.xml", this.getClass());
+		TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManagerBad");
+		PollableChannel inputTxFail = (PollableChannel) context.getBean("inputTxFailure");
+		PollableChannel inputHandlerFail = (PollableChannel) context.getBean("inputHandlerFailure");
+		PollableChannel output = (PollableChannel) context.getBean("output");
+		PollableChannel errorChannel = (PollableChannel) context.getBean("errorChannel");
+		assertEquals(0, txManager.getCommitCount());
+		inputTxFail.send(new GenericMessage<>("commitFalilureTest"));
+		Message<?> errorMessage = errorChannel.receive(10000);
+		assertNotNull(errorMessage);
+		Object payload = errorMessage.getPayload();
+		assertEquals(MessagingException.class, payload.getClass());
+		MessagingException messagingException = (MessagingException) payload;
+		assertEquals(IllegalTransactionStateException.class, messagingException.getCause().getClass());
+		assertNotNull(messagingException.getFailedMessage());
+		assertNotNull(output.receive(0));
+		assertEquals(0, txManager.getCommitCount());
+
+		inputHandlerFail.send(new GenericMessage<>("handlerFalilureTest"));
+		errorMessage = errorChannel.receive(10000);
+		assertNotNull(errorMessage);
+		payload = errorMessage.getPayload();
+		assertEquals(MessageHandlingException.class, payload.getClass());
+		messagingException = (MessageHandlingException) payload;
+		assertEquals(RuntimeException.class, messagingException.getCause().getClass());
+		assertNotNull(messagingException.getFailedMessage());
+		assertNull(output.receive(0));
+		assertEquals(0, txManager.getCommitCount());
+
+		context.close();
+	}
+
 	public static class SampleAdvice implements MethodInterceptor {
+
 		@Override
 		public Object invoke(MethodInvocation invocation) throws Throwable {
 			return invocation.proceed();
 		}
+
+	}
+
+	@SuppressWarnings("serial")
+	public static class FailingCommitTransactionManager extends TestTransactionManager {
+
+		@Override
+		protected void doCommit(DefaultTransactionStatus status) throws TransactionException {
+			throw new IllegalTransactionStateException("intentional test commit failure");
+		}
+
 	}
 
 }

spring-integration-core/src/test/java/org/springframework/integration/dispatcher/transactionFailureTests.xml

@@ -0,0 +1,51 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+			 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+			 xmlns:beans="http://www.springframework.org/schema/beans"
+			 xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration.xsd">
+
+	<channel id="inputTxFailure">
+		<queue capacity="1"/>
+	</channel>
+
+	<channel id="inputHandlerFailure">
+		<queue capacity="1"/>
+	</channel>
+
+	<channel id="output">
+		<queue capacity="1"/>
+	</channel>
+
+	<channel id="errorChannel">
+		<queue capacity="1"/>
+	</channel>
+
+
+	<service-activator input-channel="inputTxFailure" ref="testBean"
+					   method="good" output-channel="output">
+		<poller max-messages-per-poll="1" fixed-rate="10000">
+			<transactional transaction-manager="txManagerBad"/>
+		</poller>
+	</service-activator>
+
+	<service-activator input-channel="inputHandlerFailure" ref="testBean"
+					   method="bad" output-channel="output">
+		<poller max-messages-per-poll="1" fixed-rate="10000">
+			<transactional transaction-manager="txManagerGood"/>
+		</poller>
+	</service-activator>
+
+
+	<beans:bean id="testBean"
+				class="org.springframework.integration.dispatcher.TestBean"/>
+
+	<beans:bean id="txManagerBad"
+				class="org.springframework.integration.dispatcher.PollingTransactionTests$FailingCommitTransactionManager"/>
+
+	<beans:bean id="txManagerGood"
+				class="org.springframework.integration.util.TestTransactionManager"/>
+
+</beans:beans>

spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingEndpointStub.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -17,11 +17,13 @@
 package org.springframework.integration.endpoint;
 
 import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.scheduling.support.PeriodicTrigger;
 
 /**
  * @author Jonas Partner
  * @author Gary Russell
+ * @author Andreas Baer
  */
 public class PollingEndpointStub extends AbstractPollingEndpoint {
 
@@ -31,11 +33,21 @@ public class PollingEndpointStub extends AbstractPollingEndpoint {
 
 	@Override
 	protected void handleMessage(Message<?> message) {
+		throw new RuntimeException("intentional test failure");
 	}
 
 	@Override
 	protected Message<?> receiveMessage() {
-		throw new RuntimeException("intentional test failure");
+		return new GenericMessage<String>("test message");
 	}
 
+	@Override
+	protected Object getResourceToBind() {
+		return this;
+	}
+
+	@Override
+	protected String getResourceKey() {
+		return "PollingEndpointStub";
+	}
 }