Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1341: Move Tx Synch Cleanup to a finally block

Browse Source 
Resolves https://github.com/spring-projects/spring-amqp/issues/1341

For more context see the issue description.

If commit or rollback fails during synchronization `afterCompletion`
cleanup code was not called.

**cherry-pick to 2.2.x**
This commit is contained in:
Gary Russell
2021-06-02 12:02:42 -04:00
committed by Artem Bilan
parent c5ee02d851
commit 3b8fc2a7c2
3 changed files with 191 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java
View File

@@ -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

61
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateTests.java
View File

@@ -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 {

119
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerContainerTxSynchTests.java Normal file
View File

@@ -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> consumer = new AtomicReference<>();
AtomicReference<String> 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> 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();
}
}