From a7e86ba3c1c528f9984875543f0a976b6f34fb7d Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 11 Aug 2014 16:20:38 -0400 Subject: [PATCH] AMQP-421: Fix Memory Leak with Publisher Confirms 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 --- .../amqp/rabbit/core/RabbitTemplate.java | 6 ++-- ...atePublisherCallbacksIntegrationTests.java | 29 ++++++++++++------- src/reference/docbook/amqp.xml | 6 ++++ 3 files changed, 28 insertions(+), 13 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java index 861abaac..f32b0c44 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java @@ -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 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()); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java index 541942a0..bdb0a7dc 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java @@ -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() { + template.execute(new ChannelCallback() { // 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()); } diff --git a/src/reference/docbook/amqp.xml b/src/reference/docbook/amqp.xml index fb756627..89312661 100644 --- a/src/reference/docbook/amqp.xml +++ b/src/reference/docbook/amqp.xml @@ -524,6 +524,12 @@ public AmqpTemplate rabbitTemplate(); by calling setConfirmCallback(ConfirmCallback callback). The callback must implement this method: + + 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 channelCacheSize to a large enough value so that the channel on which a + message is published is returned to the cache instead of being closed. + The CorrelationData is an object supplied by the client when sending the