Apply Nullability into AMQP module

Related to: #10083

* Add `@org.jspecify.annotations.NullMarked`  to all packages of SI-AMQP module
* Fix raised concerns about `@Nullable`, including tests
* Propagate `@Nullable` to some SI-core classes where that was concerned from the AMQP classes
* The change to the `RequestReplyHeaderMapper` has led to fixes in the `DefaultSoapHeaderMapper`
and its usage in the `AbstractWebServiceOutboundGateway`
* Fix Javadocs in the `AvroTestClass1` & `AvroTestClass2` to avoid compilation warning
even if they are claimed to be generated

Author: artembilan

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractAmqpChannel.java

@@ -16,6 +16,10 @@
 
 package org.springframework.integration.amqp.channel;
 
+import java.util.Objects;
+
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.amqp.core.AmqpAdmin;
 import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.amqp.core.MessageDeliveryMode;
@@ -53,21 +57,21 @@ public abstract class AbstractAmqpChannel extends AbstractMessageChannel impleme
 
 	private final AmqpTemplate amqpTemplate;
 
-	private final RabbitTemplate rabbitTemplate;
+	private final @Nullable RabbitTemplate rabbitTemplate;
 
 	private final AmqpHeaderMapper outboundHeaderMapper;
 
 	private final AmqpHeaderMapper inboundHeaderMapper;
 
-	private AmqpAdmin admin;
+	private @Nullable AmqpAdmin admin;
 
-	private ConnectionFactory connectionFactory;
+	private @Nullable ConnectionFactory connectionFactory;
 
 	private boolean extractPayload;
 
 	private boolean loggingEnabled = true;
 
-	private MessageDeliveryMode defaultDeliveryMode;
+	private @Nullable MessageDeliveryMode defaultDeliveryMode;
 
 	private boolean headersMappedLast;
 
@@ -149,8 +153,6 @@ public void setDefaultDeliveryMode(MessageDeliveryMode defaultDeliveryMode) {
 	public void setExtractPayload(boolean extractPayload) {
 		if (extractPayload) {
 			Assert.isTrue(this.rabbitTemplate != null, "amqpTemplate must be a RabbitTemplate for 'extractPayload'");
-			Assert.state(this.outboundHeaderMapper != null && this.inboundHeaderMapper != null,
-					"'extractPayload' requires both inbound and outbound header mappers");
 		}
 		this.extractPayload = extractPayload;
 	}
@@ -211,6 +213,7 @@ protected AmqpTemplate getAmqpTemplate() {
 	}
 
 	protected RabbitTemplate getRabbitTemplate() {
+		Assert.notNull(this.rabbitTemplate, "The 'RabbitTemplate' must be provided.");
 		return this.rabbitTemplate;
 	}
 
@@ -222,11 +225,11 @@ protected final void setConnectionFactory(ConnectionFactory connectionFactory) {
 		this.connectionFactory = connectionFactory;
 	}
 
-	protected AmqpAdmin getAdmin() {
+	protected @Nullable AmqpAdmin getAdmin() {
 		return this.admin;
 	}
 
-	protected ConnectionFactory getConnectionFactory() {
+	protected @Nullable ConnectionFactory getConnectionFactory() {
 		return this.connectionFactory;
 	}
 
@@ -251,8 +254,8 @@ public void destroy() {
 	protected boolean doSend(Message<?> message, long timeout) {
 		if (this.extractPayload) {
 			this.amqpTemplate.send(getExchangeName(), getRoutingKey(), MappingUtils.mapMessage(message,
-					this.rabbitTemplate.getMessageConverter(), this.outboundHeaderMapper, this.defaultDeliveryMode,
-					this.headersMappedLast));
+					getRabbitTemplate().getMessageConverter(), Objects.requireNonNull(this.outboundHeaderMapper),
+					this.defaultDeliveryMode, this.headersMappedLast));
 		}
 		else {
 			this.amqpTemplate.convertAndSend(getExchangeName(), getRoutingKey(), message);
@@ -261,7 +264,7 @@ protected boolean doSend(Message<?> message, long timeout) {
 	}
 
 	@Override
-	public void onCreate(Connection connection) {
+	public void onCreate(@Nullable Connection connection) {
 		doDeclares();
 	}
 

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/AbstractSubscribableAmqpChannel.java

@@ -20,6 +20,7 @@
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.amqp.AmqpConnectException;
 import org.springframework.amqp.core.AmqpTemplate;
@@ -28,6 +29,7 @@
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
+import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.amqp.support.converter.SimpleMessageConverter;
 import org.springframework.integration.MessageDispatchingException;
@@ -39,7 +41,6 @@
 import org.springframework.integration.support.MessageBuilderFactory;
 import org.springframework.integration.support.management.ManageableSmartLifecycle;
 import org.springframework.messaging.Message;
-import org.springframework.messaging.MessageDeliveryException;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.util.Assert;
@@ -60,11 +61,12 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel
 
 	private final AbstractMessageListenerContainer container;
 
+	@SuppressWarnings("NullAway.Init")
 	private volatile AbstractDispatcher dispatcher;
 
 	private final boolean isPubSub;
 
-	private volatile Integer maxSubscribers;
+	private volatile @Nullable Integer maxSubscribers;
 
 	private volatile boolean declared;
 
@@ -151,8 +153,9 @@ protected AbstractSubscribableAmqpChannel(String channelName,
 	 */
 	public void setMaxSubscribers(int maxSubscribers) {
 		this.maxSubscribers = maxSubscribers;
-		if (this.dispatcher != null) {
-			this.dispatcher.setMaxSubscribers(this.maxSubscribers);
+		AbstractDispatcher dispatcherToUse = this.dispatcher;
+		if (dispatcherToUse != null) {
+			dispatcherToUse.setMaxSubscribers(maxSubscribers);
 		}
 	}
 
@@ -169,14 +172,18 @@ public boolean unsubscribe(MessageHandler handler) {
 	@Override
 	public void onInit() {
 		super.onInit();
-		this.dispatcher = this.createDispatcher();
-		if (this.maxSubscribers == null) {
-			this.maxSubscribers =
+		this.dispatcher = createDispatcher();
+		Integer maxSubscribersToCheck = this.maxSubscribers;
+		if (maxSubscribersToCheck == null) {
+			int newMaxSubscribers =
 					this.isPubSub
 							? getIntegrationProperties().getChannelsMaxBroadcastSubscribers()
 							: getIntegrationProperties().getChannelsMaxUnicastSubscribers();
+			setMaxSubscribers(newMaxSubscribers);
+		}
+		else {
+			this.dispatcher.setMaxSubscribers(maxSubscribersToCheck);
 		}
-		setMaxSubscribers(this.maxSubscribers);
 		String queue = obtainQueueName(this.channelName);
 		this.container.setQueueNames(queue);
 		MessageConverter converter =
@@ -199,17 +206,17 @@ public void onInit() {
 
 	@Override
 	public boolean isAutoStartup() {
-		return (this.container != null) && this.container.isAutoStartup();
+		return this.container.isAutoStartup();
 	}
 
 	@Override
 	public int getPhase() {
-		return (this.container != null) ? this.container.getPhase() : 0;
+		return this.container.getPhase();
 	}
 
 	@Override
 	public boolean isRunning() {
-		return (this.container != null) && this.container.isRunning();
+		return this.container.isRunning();
 	}
 
 	@Override
@@ -224,22 +231,20 @@ public void start() {
 						"Postponed to the next connection create...");
 			}
 		}
-		if (this.container != null) {
-			this.container.start();
-		}
+		this.container.start();
 	}
 
 	@Override
 	public void stop() {
-		if (this.container != null) {
+		if (isRunning()) {
 			this.container.stop();
 			this.declared = false;
 		}
 	}
 
 	@Override
 	public void stop(Runnable callback) {
-		if (this.container != null) {
+		if (isRunning()) {
 			this.container.stop(callback);
 			this.declared = false;
 		}
@@ -251,10 +256,8 @@ public void stop(Runnable callback) {
 	@Override
 	public void destroy() {
 		super.destroy();
-		if (this.container != null) {
-			this.container.destroy();
-			this.declared = false;
-		}
+		this.container.destroy();
+		this.declared = false;
 	}
 
 	protected abstract AbstractDispatcher createDispatcher();
@@ -301,25 +304,18 @@ public void onMessage(org.springframework.amqp.core.Message message) {
 				this.dispatcher.dispatch(messageToSend);
 			}
 			catch (MessageDispatchingException e) {
-				String exceptionMessage = e.getMessage() + " for amqp-channel '"
-						+ this.channel.getFullChannelName() + "'.";
-				if (this.isPubSub) {
-					// log only for backwards compatibility with pub/sub
-					if (this.logger.isWarnEnabled()) {
-						this.logger.warn(exceptionMessage, e);
-					}
-				}
-				else {
-					throw new MessageDeliveryException(messageToSend, exceptionMessage, e);
-				}
+				String exceptionMessage =
+						e.getMessage() + " for amqp-channel '" + this.channel.getFullChannelName() + "'.";
+
+				throw new ListenerExecutionFailedException(exceptionMessage, e, message);
 			}
 		}
 
 		private Message<Object> buildMessage(org.springframework.amqp.core.Message message, Object converted) {
 			AbstractIntegrationMessageBuilder<Object> messageBuilder =
 					this.messageBuilderFactory.withPayload(converted);
 			if (this.channel.isExtractPayload()) {
-				Map<String, Object> headers =
+				Map<String, @Nullable Object> headers =
 						this.inboundHeaderMapper.toHeadersFromRequest(message.getMessageProperties());
 				messageBuilder.copyHeaders(headers);
 			}

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PointToPointSubscribableAmqpChannel.java

@@ -39,6 +39,7 @@
  */
 public class PointToPointSubscribableAmqpChannel extends AbstractSubscribableAmqpChannel {
 
+	@SuppressWarnings("NullAway.Init")
 	private volatile Queue queue;
 
 	/**

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java

@@ -57,9 +57,10 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
 
 	private final String channelName;
 
+	@SuppressWarnings("NullAway.Init")
 	private Queue queue;
 
-	private CounterFacade receiveCounter;
+	private @Nullable CounterFacade receiveCounter;
 
 	private volatile int executorInterceptorsSize;
 
@@ -148,19 +149,16 @@ protected void doDeclares() {
 	}
 
 	@Override
-	@Nullable
-	public Message<?> receive() {
+	public @Nullable Message<?> receive() {
 		return doReceive(null);
 	}
 
 	@Override
-	@Nullable
-	public Message<?> receive(long timeout) {
+	public @Nullable Message<?> receive(long timeout) {
 		return doReceive(timeout);
 	}
 
-	@Nullable
-	protected Message<?> doReceive(Long timeout) {
+	protected @Nullable Message<?> doReceive(@Nullable Long timeout) {
 		ChannelInterceptorList interceptorList = getIChannelInterceptorList();
 		Deque<ChannelInterceptor> interceptorStack = null;
 		AtomicBoolean counted = new AtomicBoolean();
@@ -194,7 +192,7 @@ protected Message<?> doReceive(Long timeout) {
 	}
 
 	@Nullable
-	protected Object performReceive(Long timeout) {
+	protected Object performReceive(@Nullable Long timeout) {
 		if (!this.declared) {
 			doDeclares();
 			this.declared = true;
@@ -220,7 +218,7 @@ protected Object performReceive(Long timeout) {
 
 			if (message != null) {
 				Object payload = rabbitTemplate.getMessageConverter().fromMessage(message);
-				Map<String, Object> headers = getInboundHeaderMapper()
+				Map<String, @Nullable Object> headers = getInboundHeaderMapper()
 						.toHeadersFromRequest(message.getMessageProperties());
 				return getMessageBuilderFactory()
 						.withPayload(payload)
@@ -233,7 +231,8 @@ protected Object performReceive(Long timeout) {
 		}
 	}
 
-	private Message<?> buildMessageFromResult(@Nullable Object object, boolean traceEnabled, AtomicBoolean counted) {
+	private @Nullable Message<?> buildMessageFromResult(
+			@Nullable Object object, boolean traceEnabled, AtomicBoolean counted) {
 
 		Message<?> message = null;
 		if (object != null) {
@@ -323,8 +322,7 @@ public boolean removeInterceptor(ChannelInterceptor interceptor) {
 	}
 
 	@Override
-	@Nullable
-	public ChannelInterceptor removeInterceptor(int index) {
+	public @Nullable ChannelInterceptor removeInterceptor(int index) {
 		ChannelInterceptor interceptor = super.removeInterceptor(index);
 		if (interceptor instanceof ExecutorChannelInterceptor) {
 			this.executorInterceptorsSize--;

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java

@@ -42,8 +42,10 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel
 
 	private final Queue queue = new AnonymousQueue();
 
+	@SuppressWarnings("NullAway.Init")
 	private volatile FanoutExchange exchange;
 
+	@SuppressWarnings("NullAway.Init")
 	private volatile Binding binding;
 
 	/**

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/package-info.java

@@ -1,4 +1,5 @@
 /**
  * Provides classes related to AMQP-backed channels.
  */
+@org.jspecify.annotations.NullMarked
 package org.springframework.integration.amqp.channel;