From 7770da8446f584e05b3a2c09403c7404ee982885 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 9 Sep 2015 11:08:11 -0400 Subject: [PATCH] AMQP-532: Pub.Conf. Concurrency JIRA: https://jira.spring.io/browse/AMQP-532 Previously, pending confirms were synchronized on the map value (itself a map). A concurrent modifification exception was reported in `generateNacksForPendingAcks`. It's not clear how this could happen because all uses synchronize on the map value. However, since only one thread can use a channel at a time (obtained from the CF), it is safe to simply synchronize on the channel itself rather than using a complex locking scheme. Change synchronization to the channel object; synchronize on the channel in the `RabbitTemplate.getUnconfirmed()` method. Add javadocs to `addListener` to tell users they must synchronize on the channel. In a future release, we should not expose the map - add a note to the Javadoc. Also, while testing, it was determined that an issue could occur if the channel closes after the listener is added but before the message was sent. When the channel is a `PublisherCallbackChannel` throw an exception to the called if the underlying channel closes; call the channel so it will distribute pending acks as nacks. --- .../connection/CachingConnectionFactory.java | 8 +- .../amqp/rabbit/core/RabbitTemplate.java | 2 +- .../support/PublisherCallbackChannelImpl.java | 103 +++++++++--------- ...atePublisherCallbacksIntegrationTests.java | 82 +++++++++++++- 4 files changed, 139 insertions(+), 56 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java index 50457975..1a2d04b5 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java @@ -691,6 +691,10 @@ public class CachingConnectionFactory extends AbstractConnectionFactory } try { if (this.target == null || !this.target.isOpen()) { + if (this.target instanceof PublisherCallbackChannel) { + this.target.close(); + throw new InvocationTargetException(new AmqpException("PublisherCallbackChannel is closed")); + } this.target = null; } synchronized (targetMonitor) { @@ -703,10 +707,10 @@ public class CachingConnectionFactory extends AbstractConnectionFactory catch (InvocationTargetException ex) { if (this.target == null || !this.target.isOpen()) { // Basic re-connection logic... - this.target = null; if (logger.isDebugEnabled()) { - logger.debug("Detected closed channel on exception. Re-initializing: " + target); + logger.debug("Detected closed channel on exception. Re-initializing: " + this.target); } + this.target = null; synchronized (targetMonitor) { if (this.target == null) { this.target = createBareChannel(theConnection, transactional); 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 4a248a39..bcf83adb 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 @@ -548,7 +548,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, long threshold = System.currentTimeMillis() - age; for (Entry> channelPendingConfirmEntry : this.pendingConfirms.entrySet()) { SortedMap channelPendingConfirms = channelPendingConfirmEntry.getValue(); - synchronized(channelPendingConfirms) { + synchronized(channelPendingConfirmEntry.getKey()) { // channel Iterator> iterator = channelPendingConfirms.entrySet().iterator(); PendingConfirm pendingConfirm; while (iterator.hasNext()) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java index 3207bfc4..e33ac385 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java @@ -635,29 +635,37 @@ public class PublisherCallbackChannelImpl generateNacksForPendingAcks("Channel closed by application"); } - private void generateNacksForPendingAcks(String cause) { - synchronized (this.pendingConfirms) { - for (Entry> entry : this.pendingConfirms.entrySet()) { - Listener listener = entry.getKey(); - synchronized(entry.getValue()) { - for (Entry confirmEntry : entry.getValue().entrySet()) { - try { - confirmEntry.getValue().setCause(cause); - handleNack(confirmEntry.getKey(), false); - } - catch (IOException e) { - logger.error("Error delivering Nack afterShutdown", e); - } - } - listener.removePendingConfirmsReference(this, entry.getValue()); + private synchronized void generateNacksForPendingAcks(String cause) { + for (Entry> entry : this.pendingConfirms.entrySet()) { + Listener listener = entry.getKey(); + for (Entry confirmEntry : entry.getValue().entrySet()) { + try { + confirmEntry.getValue().setCause(cause); + handleNack(confirmEntry.getKey(), false); + } + catch (IOException e) { + logger.error("Error delivering Nack afterShutdown", e); } } - this.pendingConfirms.clear(); - this.listenerForSeq.clear(); - this.listeners.clear(); + listener.removePendingConfirmsReference(this, entry.getValue()); } + if (logger.isDebugEnabled()) { + logger.debug("PendingConfirms cleared"); + } + this.pendingConfirms.clear(); + this.listenerForSeq.clear(); + this.listeners.clear(); } + /** + * Add the listener and return the internal map of pending confirmations for that listener. + * Callers must synchronize on this channel object when modifying the map. + * This method will be changed in a future release to NOT expose the map. + * @param listener the listener. + * @return the internal map of pending confirmations. + * TODO: do not expose the map externally; change the {@code RabbitTemplate#getUnconfirmed(long)} + * functionality to delegate to a method here. + */ public synchronized SortedMap addListener(Listener listener) { Assert.notNull(listener, "Listener cannot be null"); if (this.listeners.size() == 0) { @@ -689,6 +697,9 @@ public class PublisherCallbackChannelImpl } } this.pendingConfirms.remove(listener); + if (logger.isDebugEnabled()) { + logger.debug("Removed listener " + listener); + } return result; } @@ -711,47 +722,40 @@ public class PublisherCallbackChannelImpl this.processAck(seq, false, multiple); } - private void processAck(long seq, boolean ack, boolean multiple) { + private synchronized void processAck(long seq, boolean ack, boolean multiple) { if (multiple) { /* * Piggy-backed ack - extract all Listeners for this and earlier * sequences. Then, for each Listener, handle each of it's acks. * Finally, remove the sequences from listenerForSeq. */ - synchronized(this.pendingConfirms) { - Map involvedListeners = this.listenerForSeq.headMap(seq + 1); - // eliminate duplicates - Set listeners = new HashSet(involvedListeners.values()); - for (Listener involvedListener : listeners) { - // find all unack'd confirms for this listener and handle them - SortedMap confirmsMap = this.pendingConfirms.get(involvedListener); - if (confirmsMap != null) { - synchronized(confirmsMap) { - Map confirms = confirmsMap.headMap(seq + 1); - Iterator> iterator = confirms.entrySet().iterator(); - while (iterator.hasNext()) { - Entry entry = iterator.next(); - PendingConfirm value = entry.getValue(); - iterator.remove(); - doHandleConfirm(ack, involvedListener, value); - } - } + Map involvedListeners = this.listenerForSeq.headMap(seq + 1); + // eliminate duplicates + Set listeners = new HashSet(involvedListeners.values()); + for (Listener involvedListener : listeners) { + // find all unack'd confirms for this listener and handle them + SortedMap confirmsMap = this.pendingConfirms.get(involvedListener); + if (confirmsMap != null) { + Map confirms = confirmsMap.headMap(seq + 1); + Iterator> iterator = confirms.entrySet().iterator(); + while (iterator.hasNext()) { + Entry entry = iterator.next(); + PendingConfirm value = entry.getValue(); + iterator.remove(); + doHandleConfirm(ack, involvedListener, value); } } - List seqs = new ArrayList(involvedListeners.keySet()); - for (Long key : seqs) { - this.listenerForSeq.remove(key); - } + } + List seqs = new ArrayList(involvedListeners.keySet()); + for (Long key : seqs) { + this.listenerForSeq.remove(key); } } else { Listener listener = this.listenerForSeq.remove(seq); if (listener != null) { SortedMap confirmsForListener = this.pendingConfirms.get(listener); - PendingConfirm pendingConfirm = null; - synchronized (confirmsForListener) { - pendingConfirm = confirmsForListener.remove(seq); - } + PendingConfirm pendingConfirm = confirmsForListener.remove(seq); if (pendingConfirm != null) { doHandleConfirm(ack, listener, pendingConfirm); } @@ -776,12 +780,11 @@ public class PublisherCallbackChannelImpl } } - public void addPendingConfirm(Listener listener, long seq, PendingConfirm pendingConfirm) { + public synchronized void addPendingConfirm(Listener listener, long seq, PendingConfirm pendingConfirm) { SortedMap pendingConfirmsForListener = this.pendingConfirms.get(listener); - Assert.notNull(pendingConfirmsForListener, "Listener not registered"); - synchronized (pendingConfirmsForListener) { - pendingConfirmsForListener.put(seq, pendingConfirm); - } + Assert.notNull(pendingConfirmsForListener, + "Listener not registered: " + listener + " " + this.pendingConfirms.keySet()); + pendingConfirmsForListener.put(seq, pendingConfirm); this.listenerForSeq.put(seq, listener); } 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 7d648676..e5d01a7e 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 @@ -56,6 +56,7 @@ import org.junit.Test; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; +import org.springframework.amqp.AmqpException; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; @@ -342,6 +343,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); Connection mockConnection = mock(Connection.class); Channel mockChannel = mock(Channel.class); + when(mockChannel.isOpen()).thenReturn(true); when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); when(mockConnection.isOpen()).thenReturn(true); @@ -454,6 +456,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); Connection mockConnection = mock(Connection.class); Channel mockChannel = mock(Channel.class); + when(mockChannel.isOpen()).thenReturn(true); when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); when(mockConnection.isOpen()).thenReturn(true); @@ -497,6 +500,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); Connection mockConnection = mock(Connection.class); Channel mockChannel = mock(Channel.class); + when(mockChannel.isOpen()).thenReturn(true); when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); when(mockConnection.isOpen()).thenReturn(true); @@ -508,7 +512,8 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { @Override public Object answer(InvocationOnMock invocation) throws Throwable { return count.incrementAndGet(); - }}).when(mockChannel).getNextPublishSeqNo(); + } + }).when(mockChannel).getNextPublishSeqNo(); CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); ccf.setPublisherConfirms(true); @@ -542,6 +547,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); Connection mockConnection = mock(Connection.class); Channel mockChannel = mock(Channel.class); + when(mockChannel.isOpen()).thenReturn(true); when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); when(mockConnection.isOpen()).thenReturn(true); @@ -613,6 +619,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); Connection mockConnection = mock(Connection.class); Channel mockChannel = mock(Channel.class); + when(mockChannel.isOpen()).thenReturn(true); when(mockChannel.getNextPublishSeqNo()).thenReturn(1L, 2L, 3L, 4L); when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); @@ -841,14 +848,83 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { sentAll.set(true); } }); - Collection unconfirmed = template.getUnconfirmed(-1); long t1 = System.currentTimeMillis(); while (!sentAll.get() && System.currentTimeMillis() < t1 + 20000) { - unconfirmed = template.getUnconfirmed(-1); + template.getUnconfirmed(-1); } assertTrue(sentAll.get()); assertFalse(confirmed.get()); } + // AMQP-532 ConcurrentModificationException + @Test + public void testPublisherConfirmCloseConcurrency() throws Exception { + ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); + Connection mockConnection = mock(Connection.class); + Channel mockChannel1 = mock(Channel.class); + final AtomicLong seq1 = new AtomicLong(); + doAnswer(new Answer() { + + @Override + public Long answer(InvocationOnMock invocation) throws Throwable { + return seq1.incrementAndGet(); + } + }).when(mockChannel1).getNextPublishSeqNo(); + + Channel mockChannel2 = mock(Channel.class); + when(mockChannel2.isOpen()).thenReturn(true); + final AtomicLong seq2 = new AtomicLong(); + doAnswer(new Answer() { + + @Override + public Long answer(InvocationOnMock invocation) throws Throwable { + return seq2.incrementAndGet(); + } + }).when(mockChannel2).getNextPublishSeqNo(); + + when(mockConnectionFactory.newConnection((ExecutorService) null)).thenReturn(mockConnection); + when(mockConnection.isOpen()).thenReturn(true); + when(mockConnection.createChannel()).thenReturn(mockChannel1, mockChannel2); + + CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory); + ccf.setPublisherConfirms(true); + final RabbitTemplate template = new RabbitTemplate(ccf); + + final AtomicBoolean confirmed = new AtomicBoolean(); + template.setConfirmCallback(new ConfirmCallback() { + + @Override + public void confirm(CorrelationData correlationData, boolean ack, String cause) { + confirmed.set(true); + } + }); + ExecutorService exec = Executors.newSingleThreadExecutor(); + final AtomicInteger sent = new AtomicInteger(); + doAnswer(new Answer(){ + + @Override + public Boolean answer(InvocationOnMock invocation) throws Throwable { + boolean closed = sent.incrementAndGet() < 100; + System.out.println(closed); + return closed; + } + }).when(mockChannel1).isOpen(); + final CountDownLatch sentAll = new CountDownLatch(1); + exec.execute(new Runnable() { + + @Override + public void run() { + for (int i = 0; i < 1000; i++) { + try { + template.convertAndSend(ROUTE, (Object) "message", new CorrelationData("abc")); + } + catch (AmqpException e) {} + } + sentAll.countDown(); + } + }); + assertTrue(sentAll.await(10, TimeUnit.SECONDS)); + assertTrue(confirmed.get()); + } }