Fix AMQP MessageSource Tests

https://build.spring.io/browse/INT-MASTER-983/

Race between the poller and template to get the requeued message.

Also add trace logging to the acknowledgment.

**cherry-pick to 5.0.x**

# Conflicts:
#	spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/AmqpMessageSourceIntegrationTests.java

Author: garyrussell

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpMessageSource.java

@@ -20,6 +20,9 @@
 import java.nio.charset.StandardCharsets;
 import java.util.Map;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.connection.Connection;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
@@ -199,6 +202,8 @@ public AcknowledgmentCallback createCallback(AmqpAckInfo info) {
 
 	public static class AmqpAckCallback implements AcknowledgmentCallback {
 
+		private static Log logger = LogFactory.getLog(AmqpAckCallback.class);
+
 		private final AmqpAckInfo ackInfo;
 
 		private boolean acknowledged;
@@ -235,6 +240,9 @@ public boolean isAutoAck() {
 		@Override
 		public void acknowledge(Status status) {
 			Assert.notNull(status, "'status' cannot be null");
+			if (logger.isTraceEnabled()) {
+				logger.trace("acknowledge(" + status.name() + ") for " + this);
+			}
 			try {
 				long deliveryTag = this.ackInfo.getGetResponse().getEnvelope().getDeliveryTag();
 				switch (status) {
@@ -264,6 +272,12 @@ public void acknowledge(Status status) {
 			}
 		}
 
+		@Override
+		public String toString() {
+			return "AmqpAckCallback [ackInfo=" + this.ackInfo + ", acknowledged=" + this.acknowledged
+					+ ", autoAckEnabled=" + this.autoAckEnabled + "]";
+		}
+
 	}
 
 	/**
@@ -302,6 +316,12 @@ public GetResponse getGetResponse() {
 			return this.getResponse;
 		}
 
+		@Override
+		public String toString() {
+			return "AmqpAckInfo [connection=" + this.connection + ", channel=" + this.channel + ", transacted="
+					+ this.transacted + ", getResponse=" + this.getResponse + "]";
+		}
+
 	}
 
 }

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/AmqpMessageSourceIntegrationTests.java

@@ -58,6 +58,7 @@
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.dsl.Pollers;
+import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.integration.support.AcknowledgmentCallback;
 import org.springframework.integration.support.AcknowledgmentCallback.Status;
 import org.springframework.messaging.handler.annotation.Header;
@@ -132,8 +133,9 @@ public void testImplicitNackThenAck() throws Exception {
 		assertThat(this.config.fromDsl, equalTo("bar"));
 		assertThat(this.config.fromInterceptedSource, equalTo("BAZ"));
 		assertNull(template.receive(NOAUTOACK_QUEUE));
+		assertThat(this.config.requeueLatch.getCount(), equalTo(1L));
 		this.config.callback.acknowledge(Status.REQUEUE);
-		assertNotNull(template.receive(NOAUTOACK_QUEUE, 10_000));
+		assertTrue(this.config.requeueLatch.await(10, TimeUnit.SECONDS));
 	}
 
 	@Configuration
@@ -142,6 +144,8 @@ public static class Config {
 
 		private final CountDownLatch latch = new CountDownLatch(5);
 
+		private final CountDownLatch requeueLatch = new CountDownLatch(2);
+
 		private volatile String received;
 
 		private volatile Object fromDsl;
@@ -174,7 +178,18 @@ public void in(String in) {
 		@InboundChannelAdapter(channel = "noAutoAck", poller = @Poller(fixedDelay = "100"), autoStartup = "false")
 		@Bean
 		public MessageSource<?> noAutoAckSource() {
-			return new AmqpMessageSource(connectionFactory(), NOAUTOACK_QUEUE);
+			return new AmqpMessageSource(connectionFactory(), NOAUTOACK_QUEUE) {
+
+				@Override
+				protected AbstractIntegrationMessageBuilder<Object> doReceive() {
+					AbstractIntegrationMessageBuilder<Object> builder = super.doReceive();
+					if (builder != null) {
+						Config.this.requeueLatch.countDown();
+					}
+					return builder;
+				}
+
+			};
 		}
 
 		@ServiceActivator(inputChannel = "noAutoAck")