Sonar: AMQP Fixes

- critical smells
- also import in IMHA

* Polishing

build.gradle

@@ -131,14 +131,14 @@ subprojects { subproject ->
 		romeToolsVersion = '1.9.0'
 		servletApiVersion = '4.0.0'
 		smackVersion = '4.3.1'
-		springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.1.0.RELEASE'
+		springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.1.2.BUILD-SNAPSHOT'
 		springDataJpaVersion = '2.1.2.RELEASE'
 		springDataMongoVersion = '2.1.2.RELEASE'
 		springDataRedisVersion = '2.1.2.RELEASE'
 		springGemfireVersion = '2.1.2.RELEASE'
 		springSecurityVersion = '5.1.1.RELEASE'
 		springRetryVersion = '1.2.2.RELEASE'
-		springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.1.2.RELEASE'
+		springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.1.3.BUILD-SNAPSHOT'
 		springWsVersion = '3.0.3.RELEASE'
 		tomcatVersion = "9.0.12"
 		xmlUnitVersion = '1.6'

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpPollableMessageChannelSpec.java

@@ -27,6 +27,7 @@ import org.springframework.integration.amqp.channel.AbstractAmqpChannel;
 import org.springframework.integration.amqp.config.AmqpChannelFactoryBean;
 import org.springframework.integration.amqp.support.AmqpHeaderMapper;
 import org.springframework.integration.dsl.MessageChannelSpec;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 /**
@@ -62,7 +63,7 @@ public class AmqpPollableMessageChannelSpec<S extends AmqpPollableMessageChannel
 	}
 
 	@Override
-	protected S id(String id) {
+	protected S id(@Nullable String id) {
 		this.amqpChannelFactoryBean.setBeanName(id);
 		return super.id(id);
 	}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java

@@ -18,6 +18,7 @@ package org.springframework.integration.amqp.outbound;
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.UUID;
 
 import org.springframework.amqp.core.MessageDeliveryMode;
 import org.springframework.amqp.rabbit.connection.Connection;
@@ -54,6 +55,8 @@ import org.springframework.util.StringUtils;
 public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
 		implements Lifecycle {
 
+	private static final UUID NO_ID = new UUID(0L, 0L);
+
 	private String exchangeName;
 
 	private String routingKey;
@@ -472,7 +475,11 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
 	protected CorrelationData generateCorrelationData(Message<?> requestMessage) {
 		CorrelationData correlationData = null;
 		if (this.correlationDataGenerator != null) {
-			correlationData = new CorrelationDataWrapper(requestMessage.getHeaders().getId().toString(),
+			UUID messageId = requestMessage.getHeaders().getId();
+			if (messageId == null) {
+				messageId = NO_ID;
+			}
+			correlationData = new CorrelationDataWrapper(messageId.toString(),
 					this.correlationDataGenerator.processMessage(requestMessage), requestMessage);
 		}
 		return correlationData;

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java

@@ -37,6 +37,7 @@ import static org.mockito.Mockito.when;
 import java.io.IOException;
 import java.lang.reflect.Field;
 import java.util.List;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.commons.logging.Log;
@@ -238,7 +239,8 @@ public class AmqpOutboundChannelAdapterParserTests {
 		Channel mockChannel = mock(Channel.class);
 
 		when(connectionFactory.createConnection()).thenReturn(mockConnection);
-		PublisherCallbackChannelImpl publisherCallbackChannel = new PublisherCallbackChannelImpl(mockChannel);
+		PublisherCallbackChannelImpl publisherCallbackChannel = new PublisherCallbackChannelImpl(mockChannel,
+				mock(ExecutorService.class));
 		when(mockConnection.createChannel(false)).thenReturn(publisherCallbackChannel);
 
 		MessageChannel requestChannel = context.getBean("toRabbitOnlyWithTemplateChannel", MessageChannel.class);
@@ -255,7 +257,8 @@ public class AmqpOutboundChannelAdapterParserTests {
 		Channel mockChannel = mock(Channel.class);
 
 		when(connectionFactory.createConnection()).thenReturn(mockConnection);
-		PublisherCallbackChannelImpl publisherCallbackChannel = new PublisherCallbackChannelImpl(mockChannel);
+		PublisherCallbackChannelImpl publisherCallbackChannel = new PublisherCallbackChannelImpl(mockChannel,
+				mock(ExecutorService.class));
 		when(mockConnection.createChannel(false)).thenReturn(publisherCallbackChannel);
 
 		MessageChannel requestChannel = context.getBean("withDefaultAmqpTemplateExchangeAndRoutingKey",
@@ -272,7 +275,8 @@ public class AmqpOutboundChannelAdapterParserTests {
 		Channel mockChannel = mock(Channel.class);
 
 		when(connectionFactory.createConnection()).thenReturn(mockConnection);
-		PublisherCallbackChannelImpl publisherCallbackChannel = new PublisherCallbackChannelImpl(mockChannel);
+		PublisherCallbackChannelImpl publisherCallbackChannel = new PublisherCallbackChannelImpl(mockChannel,
+				mock(ExecutorService.class));
 		when(mockConnection.createChannel(false)).thenReturn(publisherCallbackChannel);
 
 		MessageChannel requestChannel = context.getBean("overrideTemplateAttributesToEmpty", MessageChannel.class);

spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java

@@ -27,7 +27,6 @@ import java.util.concurrent.atomic.AtomicInteger;
 import org.springframework.integration.acks.AcknowledgmentCallback;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
-import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.support.MessageHeaderAccessor;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
@@ -74,10 +73,11 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor {
 	}
 
 	/**
-	 * Specify a list of headers which should be considered as read only
-	 * and prohibited from being populated in the message.
-	 * @param readOnlyHeaders the list of headers for {@code readOnly} mode.
-	 * Defaults to {@link MessageHeaders#ID} and {@link MessageHeaders#TIMESTAMP}.
+	 * Specify a list of headers which should be considered as read only and prohibited
+	 * from being populated in the message.
+	 * @param readOnlyHeaders the list of headers for {@code readOnly} mode. Defaults to
+	 * {@link org.springframework.messaging.MessageHeaders#ID} and
+	 * {@link org.springframework.messaging.MessageHeaders#TIMESTAMP}.
 	 * @since 4.3.2
 	 * @see #isReadOnly(String)
 	 */