Apply JSpecify Nullify to the Channel and AOT packages

cppwfs · cppwfs · commit 6a6002935bdf · 2025-07-01T11:01:56.000-04:00
diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java
@@ -20,6 +20,7 @@
 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;
@@ -322,12 +323,12 @@ public boolean removeInterceptor(ChannelInterceptor interceptor) {
 	}
 
 	@Override
-	public @Nullable ChannelInterceptor removeInterceptor(int index) {
+	public ChannelInterceptor removeInterceptor(int index) {
 		ChannelInterceptor interceptor = super.removeInterceptor(index);
 		if (interceptor instanceof ExecutorChannelInterceptor) {
 			this.executorInterceptorsSize--;
 		}
-		return interceptor;
+		return Objects.requireNonNull(interceptor);
 	}
 
 	@Override
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aot/CoreRuntimeHints.java b/spring-integration-core/src/main/java/org/springframework/integration/aot/CoreRuntimeHints.java
@@ -26,6 +26,8 @@
 import java.util.function.Supplier;
 import java.util.stream.Stream;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.aop.framework.AopProxyUtils;
 import org.springframework.aot.hint.ExecutableMode;
 import org.springframework.aot.hint.MemberCategory;
@@ -79,7 +81,7 @@
 class CoreRuntimeHints implements RuntimeHintsRegistrar {
 
 	@Override
-	public void registerHints(RuntimeHints hints, ClassLoader classLoader) {
+	public void registerHints(RuntimeHints hints, @Nullable ClassLoader classLoader) {
 		ReflectionHints reflectionHints = hints.reflection();
 		Stream.of(
 						GenericSelector.class,
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aot/package-info.java b/spring-integration-core/src/main/java/org/springframework/integration/aot/package-info.java
@@ -1,6 +1,5 @@
 /**
  * Provides classes to support Spring AOT.
  */
-@org.springframework.lang.NonNullApi
-@org.springframework.lang.NonNullFields
+@org.jspecify.annotations.NullMarked
 package org.springframework.integration.aot;
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java
@@ -59,13 +59,19 @@
 public abstract class AbstractExecutorChannel extends AbstractSubscribableChannel
 		implements ExecutorChannelInterceptorAware {
 
-	protected Executor executor; // NOSONAR
+	protected @Nullable Executor executor;
 
-	protected AbstractDispatcher dispatcher; // NOSONAR
+	/**
+	 * {@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;
 
-	protected Integer maxSubscribers; // NOSONAR
+	@Nullable
+	protected Integer maxSubscribers;
 
-	protected int executorInterceptorsSize; // NOSONAR
+	protected int executorInterceptorsSize;
 
 	public AbstractExecutorChannel(@Nullable Executor executor) {
 		this.executor = executor;
@@ -117,7 +123,6 @@ public boolean removeInterceptor(ChannelInterceptor interceptor) {
 	}
 
 	@Override
-	@Nullable
 	public ChannelInterceptor removeInterceptor(int index) {
 		ChannelInterceptor interceptor = super.removeInterceptor(index);
 		if (interceptor instanceof ExecutorChannelInterceptor) {
@@ -167,7 +172,7 @@ public void run() {
 				if (!CollectionUtils.isEmpty(interceptorStack)) {
 					triggerAfterMessageHandled(message, ex, interceptorStack);
 				}
-				if (ex instanceof MessagingException) { // NOSONAR
+				if (ex instanceof MessagingException) {
 					throw new MessagingExceptionWrapper(message, (MessagingException) ex);
 				}
 				String description = "Failed to handle " + message + " to " + this + " in " + messageHandler;
@@ -195,7 +200,7 @@ private Message<?> applyBeforeHandle(Message<?> message, Deque<ExecutorChannelIn
 							logger.debug(() -> executorInterceptor.getClass().getSimpleName()
 									+ " returned null from beforeHandle, i.e. precluding the send.");
 						}
-						triggerAfterMessageHandled(null, null, interceptorStack);
+						triggerAfterMessageHandled(message, null, interceptorStack);
 						return null;
 					}
 					interceptorStack.add(executorInterceptor);
@@ -204,13 +209,13 @@ private Message<?> applyBeforeHandle(Message<?> message, Deque<ExecutorChannelIn
 			return theMessage;
 		}
 
-		private void triggerAfterMessageHandled(@Nullable Message<?> message, @Nullable Exception ex,
+		private void triggerAfterMessageHandled(Message<?> message, @Nullable Exception ex,
 				Deque<ExecutorChannelInterceptor> interceptorStack) {
 			Iterator<ExecutorChannelInterceptor> iterator = interceptorStack.descendingIterator();
 			while (iterator.hasNext()) {
 				ExecutorChannelInterceptor interceptor = iterator.next();
 				try {
-					interceptor.afterMessageHandled(message, AbstractExecutorChannel.this, //NOSONAR
+					interceptor.afterMessageHandled(message, AbstractExecutorChannel.this,
 							this.delegate.getMessageHandler(), ex);
 				}
 				catch (Throwable ex2) { //NOSONAR
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java
@@ -24,6 +24,7 @@
 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;
@@ -95,28 +96,27 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 
 	private ObservationRegistry observationRegistry = ObservationRegistry.NOOP;
 
-	@Nullable
-	private MessageSenderObservationConvention observationConvention;
+	private @Nullable MessageSenderObservationConvention observationConvention;
 
 	private boolean shouldTrack = false;
 
 	private Class<?>[] datatypes = new Class<?>[0];
 
-	private MessageConverter messageConverter;
+	private @Nullable MessageConverter messageConverter;
 
 	private boolean loggingEnabled = true;
 
-	private MetricsCaptor metricsCaptor;
+	private @Nullable MetricsCaptor metricsCaptor;
 
-	private TimerFacade successTimer;
+	private @Nullable TimerFacade successTimer;
 
-	private TimerFacade failureTimer;
+	private @Nullable TimerFacade failureTimer;
 
-	private volatile String fullChannelName;
+	private volatile @Nullable String fullChannelName;
 
 	private volatile boolean applicationRunning;
 
-	private volatile Lifecycle applicationRunningController;
+	private volatile @Nullable Lifecycle applicationRunningController;
 
 	@Override
 	public String getComponentType() {
@@ -235,9 +235,8 @@ public boolean removeInterceptor(ChannelInterceptor interceptor) {
 	}
 
 	@Override
-	@Nullable
 	public ChannelInterceptor removeInterceptor(int index) {
-		return this.interceptors.remove(index);
+		return Objects.requireNonNull(this.interceptors.remove(index));
 	}
 
 	/**
@@ -394,14 +393,15 @@ private boolean sendWithObservation(Message<?> message, long timeout) {
 				messageToSendInternal =
 						new ErrorMessage(errorMessage.getPayload(),
 								messageToSend.getHeaders(),
-								errorMessage.getOriginalMessage());
+								Objects.requireNonNull(errorMessage.getOriginalMessage()));
 			}
 			return sendInternal(messageToSendInternal, timeout);
 		});
 		return Boolean.TRUE.equals(observe);
 	}
 
 	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);
@@ -469,6 +469,7 @@ private TimerFacade sendTimer(boolean sent) {
 	}
 
 	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())
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java
@@ -45,7 +45,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel
 
 	private int executorInterceptorsSize;
 
-	private CounterFacade receiveCounter;
+	private @Nullable CounterFacade receiveCounter;
 
 	@Override
 	public IntegrationPatternType getIntegrationPatternType() {
@@ -186,7 +186,6 @@ public boolean removeInterceptor(ChannelInterceptor interceptor) {
 	}
 
 	@Override
-	@Nullable
 	public ChannelInterceptor removeInterceptor(int index) {
 		ChannelInterceptor interceptor = super.removeInterceptor(index);
 		if (interceptor instanceof ExecutorChannelInterceptor) {
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelPurger.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelPurger.java
@@ -48,7 +48,7 @@ public class ChannelPurger {
 
 	private final QueueChannel[] channels;
 
-	private final MessageSelector selector;
+	private final @Nullable MessageSelector selector;
 
 	public ChannelPurger(QueueChannel... channels) {
 		this(null, channels);
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultHeaderChannelRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultHeaderChannelRegistry.java
@@ -67,7 +67,7 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport
 
 	private long reaperDelay;
 
-	private volatile ScheduledFuture<?> reaperScheduledFuture;
+	private volatile @Nullable ScheduledFuture<?> reaperScheduledFuture;
 
 	private volatile boolean running;
 
@@ -147,8 +147,9 @@ public void stop() {
 		this.lock.lock();
 		try {
 			this.running = false;
-			if (this.reaperScheduledFuture != null) {
-				this.reaperScheduledFuture.cancel(true);
+			ScheduledFuture<?> reaperScheduledFutureToCancel = this.reaperScheduledFuture;
+			if (reaperScheduledFutureToCancel != null) {
+				reaperScheduledFutureToCancel.cancel(true);
 				this.reaperScheduledFuture = null;
 			}
 			this.explicitlyStopped = true;
@@ -221,8 +222,9 @@ public MessageChannel channelNameToChannel(@Nullable String name) {
 	public void runReaper() {
 		this.lock.lock();
 		try {
-			if (this.reaperScheduledFuture != null) {
-				this.reaperScheduledFuture.cancel(true);
+			ScheduledFuture<?> reaperScheduledFutureToCancel = this.reaperScheduledFuture;
+			if (reaperScheduledFutureToCancel != null) {
+				reaperScheduledFutureToCancel.cancel(true);
 				this.reaperScheduledFuture = null;
 			}
 
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/DirectChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/DirectChannel.java
@@ -39,7 +39,7 @@ public class DirectChannel extends AbstractSubscribableChannel {
 
 	private final UnicastingDispatcher dispatcher = new UnicastingDispatcher();
 
-	private volatile Integer maxSubscribers;
+	private volatile @Nullable Integer maxSubscribers;
 
 	/**
 	 * Create a channel with default {@link RoundRobinLoadBalancingStrategy}.
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java
@@ -49,7 +49,7 @@
  */
 public class ExecutorChannel extends AbstractExecutorChannel {
 
-	private final LoadBalancingStrategy loadBalancingStrategy;
+	private final @Nullable LoadBalancingStrategy loadBalancingStrategy;
 
 	private Predicate<Exception> failoverStrategy = (exception) -> true;
 
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/FixedSubscriberChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/FixedSubscriberChannel.java
@@ -45,6 +45,7 @@ public final class FixedSubscriberChannel implements SubscribableChannel, BeanNa
 
 	private final MessageHandler handler;
 
+	@SuppressWarnings("NullAway.Init")
 	private String beanName;
 
 	public FixedSubscriberChannel() {
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java
@@ -16,6 +16,8 @@
 
 package org.springframework.integration.channel;
 
+import java.util.Objects;
+
 import org.jspecify.annotations.Nullable;
 
 import org.springframework.integration.context.IntegrationContextUtils;
@@ -47,7 +49,8 @@ public class MessagePublishingErrorHandler extends ErrorMessagePublisher impleme
 
 	private static final ErrorMessageStrategy DEFAULT_ERROR_MESSAGE_STRATEGY = (ex, attrs) -> {
 		if (ex instanceof MessagingExceptionWrapper) {
-			return new ErrorMessage(ex.getCause(), ((MessagingExceptionWrapper) ex).getFailedMessage());
+			return new ErrorMessage(Objects.requireNonNull(ex.getCause()),
+					Objects.requireNonNull(((MessagingExceptionWrapper) ex).getFailedMessage()));
 		}
 		else {
 			return new ErrorMessage(ex);
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/NullChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/NullChannel.java
@@ -34,6 +34,7 @@
 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".
@@ -60,13 +61,14 @@ public class NullChannel implements PollableChannel,
 
 	private boolean loggingEnabled = true;
 
+	@SuppressWarnings("NullAway.Init")
 	private String beanName;
 
-	private MetricsCaptor metricsCaptor;
+	private @Nullable MetricsCaptor metricsCaptor;
 
-	private TimerFacade successTimer;
+	private @Nullable TimerFacade successTimer;
 
-	private CounterFacade receiveCounter;
+	private @Nullable CounterFacade receiveCounter;
 
 	@Override
 	public void setBeanName(String beanName) {
@@ -163,6 +165,7 @@ public void onComplete() {
 
 	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")
@@ -176,7 +179,7 @@ private TimerFacade sendTimer() {
 	}
 
 	@Override
-	public Message<?> receive() {
+	public @Nullable Message<?> receive() {
 		if (this.loggingEnabled) {
 			LOG.debug("receive called on null channel");
 		}
@@ -185,7 +188,7 @@ public Message<?> receive() {
 	}
 
 	@Override
-	public Message<?> receive(long timeout) {
+	public @Nullable Message<?> receive(long timeout) {
 		return receive();
 	}
 
@@ -199,6 +202,7 @@ private void incrementReceiveCounter() {
 	}
 
 	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())
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/PartitionedChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/PartitionedChannel.java
@@ -16,6 +16,7 @@
 
 package org.springframework.integration.channel;
 
+import java.util.Objects;
 import java.util.concurrent.ThreadFactory;
 import java.util.function.Function;
 import java.util.function.Predicate;
@@ -65,7 +66,8 @@ public class PartitionedChannel extends AbstractExecutorChannel {
 	 * sent to this channel.
 	 */
 	public PartitionedChannel(int partitionCount) {
-		this(partitionCount, (message) -> message.getHeaders().get(IntegrationMessageHeaderAccessor.CORRELATION_ID));
+		this(partitionCount, (message) -> Objects.requireNonNull(
+				message.getHeaders().get(IntegrationMessageHeaderAccessor.CORRELATION_ID)));
 	}
 
 	/**
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java
@@ -135,7 +135,7 @@ protected boolean doSend(Message<?> message, long timeout) {
 	}
 
 	@Override
-	protected Message<?> doReceive(long timeout) {
+	protected @Nullable Message<?> doReceive(long timeout) {
 		Message<?> message = super.doReceive(timeout);
 		if (message != null) {
 			if (!this.useMessageStore) {
@@ -148,7 +148,7 @@ protected Message<?> doReceive(long timeout) {
 
 	private static final class SequenceFallbackComparator implements Comparator<Message<?>> {
 
-		private final Comparator<Message<?>> targetComparator;
+		private final @Nullable Comparator<Message<?>> targetComparator;
 
 		SequenceFallbackComparator(@Nullable Comparator<Message<?>> targetComparator) {
 			this.targetComparator = targetComparator;
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/package-info.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/package-info.java
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/ErrorMessagePublisher.java b/spring-integration-core/src/main/java/org/springframework/integration/core/ErrorMessagePublisher.java
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java
diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PollableKafkaChannel.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/channel/PollableKafkaChannel.java

Original file line number	Diff line number	Diff line change
`@@ -20,6 +20,7 @@`
`20`	`20`	`import java.util.Deque;`
`21`	`21`	`import java.util.List;`
`22`	`22`	`import java.util.Map;`
	`23`	`+import java.util.Objects;`
`23`	`24`	`import java.util.concurrent.atomic.AtomicBoolean;`
`24`	`25`
`25`	`26`	`import org.jspecify.annotations.Nullable;`
`@@ -322,12 +323,12 @@ public boolean removeInterceptor(ChannelInterceptor interceptor) {`
`322`	`323`	`}`
`323`	`324`
`324`	`325`	`@Override`
`325`		`- public @Nullable ChannelInterceptor removeInterceptor(int index) {`
	`326`	`+ public ChannelInterceptor removeInterceptor(int index) {`
`326`	`327`	`ChannelInterceptor interceptor = super.removeInterceptor(index);`
`327`	`328`	`if (interceptor instanceof ExecutorChannelInterceptor) {`
`328`	`329`	`this.executorInterceptorsSize--;`
`329`	`330`	`}`
`330`		`- return interceptor;`
	`331`	`+ return Objects.requireNonNull(interceptor);`
`331`	`332`	`}`
`332`	`333`
`333`	`334`	`@Override`