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

AMQP-421: Fix Memory Leak with Publisher Confirms

Browse Source 
JIRA: https://jira.spring.io/browse/AMQP-421

The wrong key is being used for `RabbitTemplate.pendingConfirms`. The rabbit template
stores the pending confirms map under the `ChannelProxy` whereas the
removal uses the actual `PublisherCallbackChannel` so the map
entry is not removed.

Use the correct key to store the pending callback map so that it
is removed from the template when closed.

Change a test case to use the `CachingConnectionFactory` instead of
a mock and verify the map is cleared.

**Add docs explaining that Publisher Confirms and Returns only works
with cached channels - ensure the cache size is sufficiently large.

Conflicts:
	src/reference/docbook/amqp.xml
This commit is contained in:
Gary Russell
2014-08-11 16:20:38 -04:00
committed by Artem Bilan
parent b5af780ad0
commit a7e86ba3c1
3 changed files with 28 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java
View File

@@ -40,6 +40,7 @@ import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.ReceiveAndReplyCallback;
import org.springframework.amqp.core.ReceiveAndReplyMessageCallback;
import org.springframework.amqp.core.ReplyToAddressCallback;
import org.springframework.amqp.rabbit.connection.ChannelProxy;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils;
import org.springframework.amqp.rabbit.connection.RabbitAccessor;
@@ -958,8 +959,9 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
if (channel instanceof PublisherCallbackChannel) {
PublisherCallbackChannel publisherCallbackChannel = (PublisherCallbackChannel) channel;
SortedMap<Long, PendingConfirm> pendingConfirms = publisherCallbackChannel.addListener(this);
if (!this.pendingConfirms.containsKey(channel)) {
this.pendingConfirms.put(channel, pendingConfirms);
Channel key = channel instanceof ChannelProxy ? ((ChannelProxy) channel).getTargetChannel() : channel;
if (!this.pendingConfirms.containsKey(key)) {
this.pendingConfirms.put(key, pendingConfirms);
if (logger.isDebugEnabled()) {
logger.debug("Added pending confirms for " + channel + " to map, size now " + this.pendingConfirms.size());
}

29
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java
View File

@@ -291,15 +291,24 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
public void testPublisherConfirmNotReceivedMultiThreads() throws Exception {
ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class);
Connection mockConnection = mock(Connection.class);
Channel mockChannel = mock(Channel.class);
Channel mockChannel1 = mock(Channel.class);
Channel mockChannel2 = mock(Channel.class);
Channel mockChannel3 = mock(Channel.class);
when(mockChannel1.isOpen()).thenReturn(true);
when(mockChannel2.isOpen()).thenReturn(true);
when(mockChannel3.isOpen()).thenReturn(true);
when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection);
when(mockConnection.isOpen()).thenReturn(true);
PublisherCallbackChannelImpl channel1 = new PublisherCallbackChannelImpl(mockChannel);
PublisherCallbackChannelImpl channel2 = new PublisherCallbackChannelImpl(mockChannel);
when(mockConnection.createChannel()).thenReturn(channel1).thenReturn(channel2);
PublisherCallbackChannelImpl channel1 = new PublisherCallbackChannelImpl(mockChannel1);
PublisherCallbackChannelImpl channel2 = new PublisherCallbackChannelImpl(mockChannel2);
PublisherCallbackChannelImpl channel3 = new PublisherCallbackChannelImpl(mockChannel3);
when(mockConnection.createChannel()).thenReturn(channel1).thenReturn(channel2).thenReturn(channel3);
final RabbitTemplate template = new RabbitTemplate(new SingleConnectionFactory(mockConnectionFactory));
CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory);
ccf.setPublisherConfirms(true);
ccf.setChannelCacheSize(3);
final RabbitTemplate template = new RabbitTemplate(ccf);
final AtomicBoolean confirmed = new AtomicBoolean();
template.setConfirmCallback(new ConfirmCallback() {
@@ -318,7 +327,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
@Override
public void run() {
template.execute(new ChannelCallback<Object>() {
template.execute(new ChannelCallback<Object>() { // channel x
@Override
public Object doInRabbit(Channel channel) throws Exception {
try {
@@ -326,7 +335,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
template.doSend(channel, "", ROUTE,
template.doSend(channel, "", ROUTE, // channel y
new SimpleMessageConverter().toMessage("message", new MessageProperties()),
new CorrelationData("def"));
threadSentLatch.countDown();
@@ -337,7 +346,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
});
// Thread 2
template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc"));
template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc")); // channel z
threadLatch.countDown();
assertTrue(threadSentLatch.await(5, TimeUnit.SECONDS));
Thread.sleep(5);
@@ -353,9 +362,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
DirectFieldAccessor dfa = new DirectFieldAccessor(template);
Map<?, ?> pendingConfirms = (Map<?, ?>) dfa.getPropertyValue("pendingConfirms");
assertEquals(2, pendingConfirms.size());
channel1.close();
assertEquals(1, pendingConfirms.size());
channel2.close();
ccf.destroy();
assertEquals(0, pendingConfirms.size());
}

6
src/reference/docbook/amqp.xml
View File

@@ -524,6 +524,12 @@ public AmqpTemplate rabbitTemplate();
by calling <code>setConfirmCallback(ConfirmCallback callback)</code>. The callback
must implement this method:
</para>
<important>
Publisher Confirms only work when the channel is cached. Otherwise, the channel is closed after the
publish operation so, by definition, cannot receive the confirmation. Be sure to set the
connection factory's <code>channelCacheSize</code> to a large enough value so that the channel on which a
message is published is returned to the cache instead of being closed.
</important>
<programlisting language="java"><![CDATA[void confirm(CorrelationData correlationData, boolean ack);]]></programlisting>
<para>
The <classname>CorrelationData</classname> is an object supplied by the client when sending the