diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java index 683b1fb635..da36e155ee 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java +++ b/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); + } } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transaction/IntegrationResourceHolderSynchronization.java b/spring-integration-core/src/main/java/org/springframework/integration/transaction/IntegrationResourceHolderSynchronization.java index a2ee9400a1..31f9979729 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transaction/IntegrationResourceHolderSynchronization.java +++ b/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 { 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; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transaction/PassThroughTransactionSynchronizationFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/transaction/PassThroughTransactionSynchronizationFactory.java new file mode 100644 index 0000000000..fc61b784e7 --- /dev/null +++ b/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; + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java index c83d571bbf..bcc5b7d941 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java +++ b/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 diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingTransactionTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingTransactionTests.java index e0299e77ce..eff6151ca7 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingTransactionTests.java +++ b/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"); + } + } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/transactionFailureTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/transactionFailureTests.xml new file mode 100644 index 0000000000..077f37a1fd --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/transactionFailureTests.xml @@ -0,0 +1,51 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingEndpointStub.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingEndpointStub.java index 72199f3e3d..a08c0fc6bd 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingEndpointStub.java +++ b/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("test message"); } + @Override + protected Object getResourceToBind() { + return this; + } + + @Override + protected String getResourceKey() { + return "PollingEndpointStub"; + } } diff --git a/src/reference/asciidoc/transactions.adoc b/src/reference/asciidoc/transactions.adoc index e56ef0219e..0b3646f070 100644 --- a/src/reference/asciidoc/transactions.adoc +++ b/src/reference/asciidoc/transactions.adoc @@ -165,6 +165,12 @@ For example, the `MongoDbMessageSource` provides the _#mongoTemplate_ variable w To enable the feature for a particular poller, you provide a reference to the `TransactionSynchronizationFactory` on the poller's element using the _synchronization-factory_ attribute. +Starting with _version 5.0_, a new `PassThroughTransactionSynchronizationFactory` is provided which is applied by default to polling endpoints when no `TransactionSynchronizationFactory` is configured but an advice of type TransactionInterceptor exists in the advice chain. +When using any out-of-the-box `TransactionSynchronizationFactory` implementation, polling endpoints bind a polled message to the current transactional context and provide it as a `failedMessage` in a `MessagingException` if an exception is thrown after the TX advice. +When using a custom TX advice that does not implement `TransactionInterceptor`, a `PassThroughTransactionSynchronizationFactory` can be configured explicitly to achieve this behavior. +In either case, the `MessagingException` becomes the payload of the `ErrorMessage` that is sent to the `errorChannel` and the cause is the raw exception thrown by the advice. +Previously, the `ErrorMessage` had a payload that was the raw exception thrown by the advice and did not provide a reference to the `failedMessage` information, making it difficult to determine the reasons for the transaction commit problem. + To simplify configuration of these components, namespace support for the default factory has been provided. Configuration is best described using an example: diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 08bdbdc32a..3998fa912b 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -48,6 +48,10 @@ See <> for more information. When targeting POJO methods as message handlers, one of the service methods can now be marked with the `@Default` annotation to provide a fallback mechanism for non-matched conditions. See <> for more information. +A simple `PassThroughTransactionSynchronizationFactory` is provided to always store a polled message in the current transaction context. +That message is used as a `failedMessage` property of the `MessagingException` which wraps a raw exception thrown during transaction completion. +See <> for more information. + ==== JMS Changes Previously, Spring Integration JMS XML configuration used a default bean name `connectionFactory` for the JMS Connection Factory, allowing the property to be omitted from component definitions.