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.

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);

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<Object, SortedMap<Long, PendingConfirm>> channelPendingConfirmEntry : this.pendingConfirms.entrySet()) {
 				SortedMap<Long, PendingConfirm> channelPendingConfirms = channelPendingConfirmEntry.getValue();
-				synchronized(channelPendingConfirms) {
+				synchronized(channelPendingConfirmEntry.getKey()) { // channel
 					Iterator<Entry<Long, PendingConfirm>> iterator = channelPendingConfirms.entrySet().iterator();
 					PendingConfirm pendingConfirm;
 					while (iterator.hasNext()) {

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<Listener, SortedMap<Long, PendingConfirm>> entry : this.pendingConfirms.entrySet()) {
-				Listener listener = entry.getKey();
-				synchronized(entry.getValue()) {
-					for (Entry<Long, PendingConfirm> 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<Listener, SortedMap<Long, PendingConfirm>> entry : this.pendingConfirms.entrySet()) {
+			Listener listener = entry.getKey();
+			for (Entry<Long, PendingConfirm> 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 <b>must</b> 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<Long, PendingConfirm> 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<Long, Listener> involvedListeners = this.listenerForSeq.headMap(seq + 1);
-				// eliminate duplicates
-				Set<Listener> listeners = new HashSet<Listener>(involvedListeners.values());
-				for (Listener involvedListener : listeners) {
-					// find all unack'd confirms for this listener and handle them
-					SortedMap<Long, PendingConfirm> confirmsMap = this.pendingConfirms.get(involvedListener);
-					if (confirmsMap != null) {
-						synchronized(confirmsMap) {
-							Map<Long, PendingConfirm> confirms = confirmsMap.headMap(seq + 1);
-							Iterator<Entry<Long, PendingConfirm>> iterator = confirms.entrySet().iterator();
-							while (iterator.hasNext()) {
-								Entry<Long, PendingConfirm> entry = iterator.next();
-								PendingConfirm value = entry.getValue();
-								iterator.remove();
-								doHandleConfirm(ack, involvedListener, value);
-							}
-						}
+			Map<Long, Listener> involvedListeners = this.listenerForSeq.headMap(seq + 1);
+			// eliminate duplicates
+			Set<Listener> listeners = new HashSet<Listener>(involvedListeners.values());
+			for (Listener involvedListener : listeners) {
+				// find all unack'd confirms for this listener and handle them
+				SortedMap<Long, PendingConfirm> confirmsMap = this.pendingConfirms.get(involvedListener);
+				if (confirmsMap != null) {
+					Map<Long, PendingConfirm> confirms = confirmsMap.headMap(seq + 1);
+					Iterator<Entry<Long, PendingConfirm>> iterator = confirms.entrySet().iterator();
+					while (iterator.hasNext()) {
+						Entry<Long, PendingConfirm> entry = iterator.next();
+						PendingConfirm value = entry.getValue();
+						iterator.remove();
+						doHandleConfirm(ack, involvedListener, value);
 					}
 				}
-				List<Long> seqs = new ArrayList<Long>(involvedListeners.keySet());
-				for (Long key : seqs) {
-					this.listenerForSeq.remove(key);
-				}
+			}
+			List<Long> seqs = new ArrayList<Long>(involvedListeners.keySet());
+			for (Long key : seqs) {
+				this.listenerForSeq.remove(key);
 			}
 		}
 		else {
 			Listener listener = this.listenerForSeq.remove(seq);
 			if (listener != null) {
 				SortedMap<Long, PendingConfirm> 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<Long, PendingConfirm> 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);
 	}
 

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<CorrelationData> 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<Long>() {
+
+			@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<Long>() {
+
+			@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<Boolean>(){
+
+			@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());
+	}
 
 }