Adjust PR to resolve code review issues

* In code segments where performance is critical remove null checks and suppress warnings.
* Format code to improve readability
* Cleanup some Nullable's in a couple locations

Continued effort or issue: #10083

Author: cppwfs

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

@@ -20,7 +20,6 @@
 import java.util.Deque;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.jspecify.annotations.Nullable;
@@ -328,7 +327,7 @@ public ChannelInterceptor removeInterceptor(int index) {
 		if (interceptor instanceof ExecutorChannelInterceptor) {
 			this.executorInterceptorsSize--;
 		}
-		return Objects.requireNonNull(interceptor);
+		return interceptor;
 	}
 
 	@Override

spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java

@@ -61,10 +61,6 @@ public abstract class AbstractExecutorChannel extends AbstractSubscribableChanne
 
 	protected @Nullable Executor executor;
 
-	/**
-	 * {@code @SuppressWarnings} was used the dispatcher is initialized in each of the implementations in the {@link AbstractExecutorChannel}.   And each implementation
-	 * utilizes a unique implementation of the {@link AbstractDispatcher}.
-	 */
 	@SuppressWarnings("NullAway.Init")
 	protected AbstractDispatcher dispatcher;
 

spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java

@@ -24,7 +24,6 @@
 import java.util.Deque;
 import java.util.Iterator;
 import java.util.List;
-import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -236,7 +235,7 @@ public boolean removeInterceptor(ChannelInterceptor interceptor) {
 
 	@Override
 	public ChannelInterceptor removeInterceptor(int index) {
-		return Objects.requireNonNull(this.interceptors.remove(index));
+		return this.interceptors.remove(index);
 	}
 
 	/**
@@ -391,17 +390,18 @@ private boolean sendWithObservation(Message<?> message, long timeout) {
 			Message<?> messageToSendInternal = messageToSend;
 			if (message instanceof ErrorMessage errorMessage) {
 				messageToSendInternal =
-						new ErrorMessage(errorMessage.getPayload(),
+						(errorMessage.getOriginalMessage() != null) ? new ErrorMessage(errorMessage.getPayload(),
 								messageToSend.getHeaders(),
-								Objects.requireNonNull(errorMessage.getOriginalMessage()));
+								errorMessage.getOriginalMessage()) : new ErrorMessage(errorMessage.getPayload(),
+								messageToSend.getHeaders());
 			}
 			return sendInternal(messageToSendInternal, timeout);
 		});
 		return Boolean.TRUE.equals(observe);
 	}
 
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	private boolean sendWithMetrics(Message<?> message, long timeout) {
-		Assert.state(this.metricsCaptor != null, "The metricsCaptor must not be null");
 		SampleFacade sample = this.metricsCaptor.start();
 		try {
 			boolean sent = sendInternal(message, timeout);
@@ -468,8 +468,8 @@ private TimerFacade sendTimer(boolean sent) {
 		}
 	}
 
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	private TimerFacade buildSendTimer(boolean success, String exception) {
-		Assert.state(this.metricsCaptor != null, "The metricsCaptor must not be null");
 		TimerFacade timer = this.metricsCaptor.timerBuilder(SEND_TIMER_NAME)
 				.tag("type", "channel")
 				.tag("name", getComponentName() == null ? "unknown" : getComponentName())
@@ -677,7 +677,6 @@ public boolean remove(ChannelInterceptor interceptor) {
 			}
 		}
 
-		@Nullable
 		public ChannelInterceptor remove(int index) {
 			ChannelInterceptor removed = this.interceptors.remove(index);
 			if (removed != null) {

spring-integration-core/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java

@@ -16,8 +16,6 @@
 
 package org.springframework.integration.channel;
 
-import java.util.Objects;
-
 import org.jspecify.annotations.Nullable;
 
 import org.springframework.integration.context.IntegrationContextUtils;
@@ -47,10 +45,11 @@ public class MessagePublishingErrorHandler extends ErrorMessagePublisher impleme
 
 	private static final int DEFAULT_SEND_TIMEOUT = 1000;
 
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	private static final ErrorMessageStrategy DEFAULT_ERROR_MESSAGE_STRATEGY = (ex, attrs) -> {
-		if (ex instanceof MessagingExceptionWrapper) {
-			return new ErrorMessage(Objects.requireNonNull(ex.getCause()),
-					Objects.requireNonNull(((MessagingExceptionWrapper) ex).getFailedMessage()));
+		if (ex instanceof MessagingExceptionWrapper messagingExceptionWrapper) {
+			return new ErrorMessage(messagingExceptionWrapper.getCause(),
+					messagingExceptionWrapper.getFailedMessage());
 		}
 		else {
 			return new ErrorMessage(ex);
@@ -69,7 +68,7 @@ public MessagePublishingErrorHandler(DestinationResolver<MessageChannel> channel
 		setChannelResolver(channelResolver);
 	}
 
-	public void setDefaultErrorChannel(@Nullable MessageChannel defaultErrorChannel) {
+	public void setDefaultErrorChannel(MessageChannel defaultErrorChannel) {
 		setChannel(defaultErrorChannel);
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/channel/NullChannel.java

@@ -34,7 +34,6 @@
 import org.springframework.integration.support.management.metrics.TimerFacade;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.PollableChannel;
-import org.springframework.util.Assert;
 
 /**
  * A channel implementation that essentially behaves like "/dev/null".
@@ -163,9 +162,9 @@ public void onComplete() {
 		return true;
 	}
 
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	private TimerFacade sendTimer() {
 		if (this.successTimer == null) {
-			Assert.state(this.metricsCaptor != null, "metricsCaptor not must not be null");
 			this.successTimer =
 					this.metricsCaptor.timerBuilder(SEND_TIMER_NAME)
 							.tag("type", "channel")
@@ -201,8 +200,8 @@ private void incrementReceiveCounter() {
 		}
 	}
 
+	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	private CounterFacade buildReceiveCounter() {
-		Assert.state(this.metricsCaptor != null, "metricsCaptor not must not be null");
 		return this.metricsCaptor
 				.counterBuilder(RECEIVE_COUNTER_NAME)
 				.tag("name", getComponentName() == null ? "unknown" : getComponentName())

spring-integration-core/src/main/java/org/springframework/integration/channel/PartitionedChannel.java

@@ -66,8 +66,7 @@ public class PartitionedChannel extends AbstractExecutorChannel {
 	 * sent to this channel.
 	 */
 	public PartitionedChannel(int partitionCount) {
-		this(partitionCount, (message) -> Objects.requireNonNull(
-				message.getHeaders().get(IntegrationMessageHeaderAccessor.CORRELATION_ID)));
+		this(partitionCount, (message) -> Objects.requireNonNull(message.getHeaders().get(IntegrationMessageHeaderAccessor.CORRELATION_ID)));
 	}
 
 	/**

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

@@ -70,7 +70,7 @@ public final void setErrorMessageStrategy(ErrorMessageStrategy errorMessageStrat
 		this.errorMessageStrategy = errorMessageStrategy;
 	}
 
-	public final void setChannel(@Nullable MessageChannel channel) {
+	public final void setChannel(MessageChannel channel) {
 		this.channel = channel;
 	}