diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java index ede632de..f2c66994 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java @@ -330,17 +330,20 @@ public final class ConnectionFactoryUtils { @Override public void afterCompletion(int status) { - if (status == TransactionSynchronization.STATUS_COMMITTED) { - this.resourceHolder.commitAll(); + try { + if (status == TransactionSynchronization.STATUS_COMMITTED) { + this.resourceHolder.commitAll(); + } + else { + this.resourceHolder.rollbackAll(); + } } - else { - this.resourceHolder.rollbackAll(); + finally { + if (this.resourceHolder.isReleaseAfterCompletion()) { + this.resourceHolder.setSynchronizedWithTransaction(false); + } + super.afterCompletion(status); } - - if (this.resourceHolder.isReleaseAfterCompletion()) { - this.resourceHolder.setSynchronizedWithTransaction(false); - } - super.afterCompletion(status); } @Override diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateTests.java index a8cbcefc..7208ba7d 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 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,6 +33,7 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import java.io.IOException; import java.util.Collection; import java.util.Collections; import java.util.HashMap; @@ -41,6 +42,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -67,6 +69,7 @@ import org.springframework.amqp.rabbit.connection.RabbitUtils; import org.springframework.amqp.rabbit.connection.SimpleRoutingConnectionFactory; import org.springframework.amqp.rabbit.connection.SingleConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnsCallback; +import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager; import org.springframework.amqp.support.converter.SimpleMessageConverter; import org.springframework.amqp.utils.SerializationUtils; import org.springframework.amqp.utils.test.TestUtils; @@ -79,9 +82,11 @@ import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.TransactionException; import org.springframework.transaction.support.AbstractPlatformTransactionManager; import org.springframework.transaction.support.DefaultTransactionStatus; +import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.transaction.support.TransactionTemplate; import com.rabbitmq.client.AMQP; +import com.rabbitmq.client.AMQP.Tx.SelectOk; import com.rabbitmq.client.AuthenticationFailureException; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; @@ -581,6 +586,60 @@ public class RabbitTemplateTests { template2.setReturnCallback(callback); } + @Test + void resourcesClearedAfterTxFails() throws IOException, TimeoutException { + ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); + Connection mockConnection = mock(Connection.class); + Channel mockChannel = mock(Channel.class); + + given(mockConnectionFactory.newConnection(any(ExecutorService.class), anyString())).willReturn(mockConnection); + given(mockConnection.isOpen()).willReturn(true); + given(mockConnection.createChannel()).willReturn(mockChannel); + given(mockChannel.txSelect()).willReturn(mock(SelectOk.class)); + given(mockChannel.txCommit()).willThrow(IllegalStateException.class); + SingleConnectionFactory connectionFactory = new SingleConnectionFactory(mockConnectionFactory); + connectionFactory.setExecutor(mock(ExecutorService.class)); + RabbitTemplate template = new RabbitTemplate(connectionFactory); + template.setChannelTransacted(true); + TransactionTemplate tt = new TransactionTemplate(new RabbitTransactionManager(connectionFactory)); + assertThatIllegalStateException() + .isThrownBy(() -> + tt.execute(status -> { + template.convertAndSend("foo", "bar"); + return null; + })); + assertThat(TransactionSynchronizationManager.hasResource(connectionFactory)).isFalse(); + assertThatIllegalStateException() + .isThrownBy(() -> (TransactionSynchronizationManager.getSynchronizations()).isEmpty()) + .withMessage("Transaction synchronization is not active"); + } + + @Test + void resourcesClearedAfterTxFailsWithSync() throws IOException, TimeoutException { + ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); + Connection mockConnection = mock(Connection.class); + Channel mockChannel = mock(Channel.class); + + given(mockConnectionFactory.newConnection(any(ExecutorService.class), anyString())).willReturn(mockConnection); + given(mockConnection.isOpen()).willReturn(true); + given(mockConnection.createChannel()).willReturn(mockChannel); + given(mockChannel.txSelect()).willReturn(mock(SelectOk.class)); + given(mockChannel.txCommit()).willThrow(IllegalStateException.class); + SingleConnectionFactory connectionFactory = new SingleConnectionFactory(mockConnectionFactory); + connectionFactory.setExecutor(mock(ExecutorService.class)); + RabbitTemplate template = new RabbitTemplate(connectionFactory); + template.setChannelTransacted(true); + TransactionTemplate tt = new TransactionTemplate(new TestTransactionManager()); + tt.execute(status -> { + template.convertAndSend("foo", "bar"); + return null; + }); + assertThat(TransactionSynchronizationManager.hasResource(connectionFactory)).isFalse(); + assertThatIllegalStateException() + .isThrownBy(() -> (TransactionSynchronizationManager.getSynchronizations()).isEmpty()) + .withMessage("Transaction synchronization is not active"); + } + @SuppressWarnings("serial") private class TestTransactionManager extends AbstractPlatformTransactionManager { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerTxSynchTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerTxSynchTests.java new file mode 100644 index 00000000..fb097e44 --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerTxSynchTests.java @@ -0,0 +1,119 @@ +/* + * Copyright 2021 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.amqp.rabbit.listener; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIllegalStateException; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.BDDMockito.given; +import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.Mockito.mock; + +import java.io.IOException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.jupiter.api.Test; + +import org.springframework.amqp.rabbit.connection.SingleConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.transaction.support.TransactionSynchronizationManager; + +import com.rabbitmq.client.AMQP; +import com.rabbitmq.client.AMQP.Tx.SelectOk; +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.ConnectionFactory; +import com.rabbitmq.client.Consumer; +import com.rabbitmq.client.Envelope; + +/** + * @author Gary Russell + * @since 2.2.18 + * + */ +public class MessageListenerContainerTxSynchTests { + + @Test + void resourcesClearedAfterTxFails() throws IOException, TimeoutException, InterruptedException, ExecutionException { + ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); + Connection mockConnection = mock(Connection.class); + Channel mockChannel = mock(Channel.class); + + given(mockConnectionFactory.newConnection(any(ExecutorService.class), anyString())).willReturn(mockConnection); + given(mockConnection.isOpen()).willReturn(true); + given(mockConnection.createChannel()).willReturn(mockChannel); + given(mockChannel.txSelect()).willReturn(mock(SelectOk.class)); + given(mockChannel.txCommit()).willThrow(IllegalStateException.class); + AtomicReference consumer = new AtomicReference<>(); + AtomicReference tag = new AtomicReference<>(); + CountDownLatch latch1 = new CountDownLatch(1); + willAnswer(inv -> { + consumer.set(inv.getArgument(6)); + consumer.get().handleConsumeOk(inv.getArgument(2)); + tag.set(inv.getArgument(2)); + latch1.countDown(); + return null; + }).given(mockChannel).basicConsume(any(), anyBoolean(), any(), anyBoolean(), anyBoolean(), any(), any()); + SingleConnectionFactory connectionFactory = new SingleConnectionFactory(mockConnectionFactory); + connectionFactory.setExecutor(mock(ExecutorService.class)); + RabbitTemplate template = new RabbitTemplate(connectionFactory); + template.setChannelTransacted(true); + ThreadPoolTaskExecutor exec = new ThreadPoolTaskExecutor(); + exec.initialize(); + SimpleMessageListenerContainer mlc = new SimpleMessageListenerContainer(); + mlc.setConnectionFactory(connectionFactory); + mlc.setQueueNames("foo"); + mlc.setTaskExecutor(exec); + mlc.setChannelTransacted(true); + CountDownLatch latch2 = new CountDownLatch(1); + mlc.setMessageListener(msg -> { + template.convertAndSend("foo", "bar"); + latch2.countDown(); + }); + mlc.start(); + assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue(); + consumer.get().handleDelivery(tag.get(), new Envelope(1, false, "", ""), new AMQP.BasicProperties(), + new byte[0]); + assertThat(latch2.await(10, TimeUnit.SECONDS)).isTrue(); + mlc.stop(); + Future exception = exec.submit(() -> { + try { + // verify no lingering resources bound to the executor's single thread + assertThat(TransactionSynchronizationManager.hasResource(connectionFactory)).isFalse(); + assertThatIllegalStateException() + .isThrownBy(() -> (TransactionSynchronizationManager.getSynchronizations()).isEmpty()) + .withMessage("Transaction synchronization is not active"); + return null; + } + catch (Exception e) { + return e; + } + }); + assertThat(exception.get(10, TimeUnit.SECONDS)).isNull(); + exec.shutdown(); + } + +}