Fix race condition in the AMQP ChannelTests

https://build.spring.io/browse/INT-AT42SIO-502

There is a small time window when we remove the current consumer
from the local store, but there is no a new one yet.
So, we have to check the `Set`(`Map`) size before calling its `iterator`

**Cherry-pick to 4.3.x**

Conflicts:
	spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java

Author: artembilan

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java

@@ -25,6 +25,7 @@
 
 import java.util.Collection;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.TimeUnit;
 
@@ -110,19 +111,24 @@ public void handleMessage(Message<?> message) throws MessagingException {
 		assertEquals(0, TestUtils.getPropertyValue(factory, "connectionListener.delegates", Collection.class).size());
 	}
 
+	@SuppressWarnings("unchecked")
 	private void waitForNewConsumer(PublishSubscribeAmqpChannel channel, BlockingQueueConsumer consumer)
 			throws Exception {
-		BlockingQueueConsumer newConsumer = (BlockingQueueConsumer) TestUtils
-				.getPropertyValue(channel, "container.consumers", Map.class).keySet().iterator().next();
+
+		final Object consumersMonitor = TestUtils.getPropertyValue(channel, "container.consumersMonitor");
 		int n = 0;
-		boolean newConsumerIsConsuming = newConsumer != consumer && TestUtils.getPropertyValue(newConsumer,
-				"consumerTags", Map.class).size() > 0;
-		while (n++ < 100 && !newConsumerIsConsuming) {
+		while (n++ < 100) {
+			Set<BlockingQueueConsumer> consumers = TestUtils.getPropertyValue(channel, "container.consumers", Map.class).keySet();
+			synchronized (consumersMonitor) {
+				if (!consumers.isEmpty()) {
+					BlockingQueueConsumer newConsumer = consumers.iterator().next();
+					if (newConsumer != consumer && TestUtils.getPropertyValue(newConsumer,
+							"consumerTags", Map.class).size() > 0) {
+						break;
+					}
+				}
+			}
 			Thread.sleep(100);
-			newConsumer = (BlockingQueueConsumer) TestUtils
-					.getPropertyValue(channel, "container.consumers", Map.class).keySet().iterator().next();
-			newConsumerIsConsuming = newConsumer != consumer && TestUtils.getPropertyValue(newConsumer,
-					"consumerTags", Map.class).size() > 0;
 		}
 		assertTrue("Failed to restart consumer", n < 100);
 	}