Introduce NullMarked to some packages in spring-integration-core module

spring-integration-core/src/main/java/org/springframework/integration/acks/package-info.java

@@ -1,5 +1,5 @@
 /**
  * Provides classes related to message acknowledgment.
  */
-@org.springframework.lang.NonNullApi
+@org.jspecify.annotations.NullMarked
 package org.springframework.integration.acks;

spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -56,6 +56,7 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag
 
 	private boolean messageBuilderFactorySet;
 
+	@SuppressWarnings("NullAway.Init")
 	private BeanFactory beanFactory;
 
 	@Override

spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java

@@ -125,8 +125,10 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 
 	private boolean releaseStrategySet;
 
+	@SuppressWarnings("NullAway.Init")
 	private MessageChannel discardChannel;
 
+	@Nullable
 	private String discardChannelName;
 
 	private boolean sendPartialResultOnExpiry;
@@ -143,18 +145,23 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 
 	private boolean releasePartialSequences;
 
+	@Nullable
 	private Expression groupTimeoutExpression;
 
+	@Nullable
 	private List<Advice> forceReleaseAdviceChain;
 
 	private long expireTimeout;
 
+	@Nullable
 	private Duration expireDuration;
 
 	private MessageGroupProcessor forceReleaseProcessor = new ForceReleaseMessageGroupProcessor();
 
+	@SuppressWarnings("NullAway.Init")
 	private EvaluationContext evaluationContext;
 
+	@Nullable
 	private ApplicationEventPublisher applicationEventPublisher;
 
 	private boolean expireGroupsUponTimeout = true;
@@ -165,10 +172,11 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 
 	private volatile boolean running;
 
+	@Nullable
 	private BiFunction<Message<?>, String, String> groupConditionSupplier;
 
 	public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store,
-			CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) {
+			@Nullable CorrelationStrategy correlationStrategy, @Nullable ReleaseStrategy releaseStrategy) {
 
 		Assert.notNull(processor, "'processor' must not be null");
 		Assert.notNull(store, "'store' must not be null");
@@ -504,6 +512,7 @@ public MessageChannel getDiscardChannel() {
 		return this.discardChannel;
 	}
 
+	@Nullable
 	protected String getDiscardChannelName() {
 		return this.discardChannelName;
 	}
@@ -532,6 +541,7 @@ protected boolean isReleasePartialSequences() {
 		return this.releasePartialSequences;
 	}
 
+	@Nullable
 	protected Expression getGroupTimeoutExpression() {
 		return this.groupTimeoutExpression;
 	}
@@ -753,7 +763,7 @@ private void discardMessage(Message<?> message) {
 	 * @param group The group.
 	 * @param completedMessages The completed messages.
 	 */
-	protected abstract void afterRelease(MessageGroup group, Collection<Message<?>> completedMessages);
+	protected abstract void afterRelease(MessageGroup group, @Nullable Collection<Message<?>> completedMessages);
 
 	/**
 	 * Subclasses may override if special action is needed because the group was released or discarded
@@ -912,14 +922,12 @@ protected void expireGroup(Object correlationKey, MessageGroup group, Lock lock)
 	}
 
 	protected void completeGroup(Object correlationKey, MessageGroup group, Lock lock) {
-		Message<?> first = null;
-		if (group != null) {
-			first = group.getOne();
-		}
+		Message<?> first = group.getOne();
 		completeGroup(first, correlationKey, group, lock);
 	}
 
 	@SuppressWarnings("unchecked")
+	@Nullable
 	protected Collection<Message<?>> completeGroup(Message<?> message, Object correlationKey, MessageGroup group,
 			Lock lock) {
 
@@ -929,6 +937,7 @@ protected Collection<Message<?>> completeGroup(Message<?> message, Object correl
 			this.logger.debug(() -> "Completing group with correlationKey [" + correlationKey + "]");
 
 			result = this.outputProcessor.processMessageGroup(group);
+			Assert.state(result != null, "The processorMessageGroup returned a null result. Null result is not expected.");
 			if (isResultCollectionOfMessages(result)) {
 				partialSequence = (Collection<Message<?>>) result;
 			}
@@ -988,6 +997,7 @@ private static boolean isResultCollectionOfMessages(Object result) {
 		return false;
 	}
 
+	@Nullable
 	protected Object obtainGroupTimeout(MessageGroup group) {
 		if (this.groupTimeoutExpression != null) {
 			Object timeout = this.groupTimeoutExpression.getValue(this.evaluationContext, group);
@@ -1062,6 +1072,7 @@ public void purgeOrphanedGroups() {
 
 	protected static class SequenceAwareMessageGroup extends SimpleMessageGroup {
 
+		@Nullable
 		private final SimpleMessageGroup sourceGroup;
 
 		public SequenceAwareMessageGroup(MessageGroup messageGroup) {
@@ -1124,6 +1135,7 @@ private class ForceReleaseMessageGroupProcessor implements MessageGroupProcessor
 		}
 
 		@Override
+		@Nullable
 		public Object processMessageGroup(MessageGroup group) {
 			forceComplete(group);
 			return null;

spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2024 the original author or authors.
+ * Copyright 2002-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,6 +18,8 @@
 
 import java.util.Collection;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.integration.IntegrationPatternType;
 import org.springframework.integration.store.MessageGroup;
 import org.springframework.integration.store.MessageGroupStore;
@@ -95,7 +97,7 @@ protected boolean shouldSplitOutput(Iterable<?> reply) {
 	 * @param completedMessages The completed messages. Ignored in this implementation.
 	 */
 	@Override
-	protected void afterRelease(MessageGroup messageGroup, Collection<Message<?>> completedMessages) {
+	protected void afterRelease(MessageGroup messageGroup, @Nullable Collection<Message<?>> completedMessages) {
 		Object groupId = messageGroup.getGroupId();
 		MessageGroupStore messageStore = getMessageStore();
 		messageStore.completeGroup(groupId);

spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java

@@ -68,8 +68,10 @@ public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler
 
 	private final MessageGroupProcessor messageGroupProcessor;
 
+	@Nullable
 	private String discardChannelName;
 
+	@Nullable
 	private MessageChannel discardChannel;
 
 	/**
@@ -159,7 +161,7 @@ public BarrierMessageHandler(long requestTimeout, long triggerTimeout, Correlati
 	 * @since 5.4
 	 */
 	public BarrierMessageHandler(long requestTimeout, long triggerTimeout, MessageGroupProcessor outputProcessor,
-			CorrelationStrategy correlationStrategy) {
+			@Nullable CorrelationStrategy correlationStrategy) {
 
 		Assert.notNull(outputProcessor, "'messageGroupProcessor' cannot be null");
 		this.messageGroupProcessor = outputProcessor;
@@ -218,6 +220,7 @@ public IntegrationPatternType getIntegrationPatternType() {
 	}
 
 	@Override
+	@Nullable
 	protected Object handleRequestMessage(Message<?> requestMessage) {
 		Object key = this.correlationStrategy.getCorrelationKey(requestMessage);
 		if (key == null) {
@@ -247,6 +250,7 @@ protected Object handleRequestMessage(Message<?> requestMessage) {
 		return null;
 	}
 
+	@Nullable
 	private Object processRelease(Object key, Message<?> requestMessage, Message<?> releaseMessage) {
 		this.suspensions.remove(key);
 		if (releaseMessage.getPayload() instanceof Throwable) {
@@ -266,6 +270,7 @@ private Object processRelease(Object key, Message<?> requestMessage, Message<?>
 	 * @param releaseMessage the release message.
 	 * @return the result.
 	 */
+	@Nullable
 	protected Object buildResult(Object key, Message<?> requestMessage, Message<?> releaseMessage) {
 		SimpleMessageGroup group = new SimpleMessageGroup(key);
 		group.add(requestMessage);

spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageBarrier.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2024 the original author or authors.
+ * Copyright 2002-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,13 +20,16 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.core.log.LogMessage;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.integration.store.MessageGroup;
 import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.store.SimpleMessageStore;
 import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
 
 /**
  * This Endpoint serves as a barrier for messages that should not be processed yet. The decision when a message can be
@@ -58,8 +61,10 @@ public class CorrelatingMessageBarrier extends AbstractMessageHandler implements
 
 	private final MessageGroupStore store;
 
+	@Nullable
 	private CorrelationStrategy correlationStrategy;
 
+	@Nullable
 	private ReleaseStrategy releaseStrategy;
 
 	public CorrelatingMessageBarrier() {
@@ -88,7 +93,9 @@ public void setReleaseStrategy(ReleaseStrategy releaseStrategy) {
 
 	@Override
 	protected void handleMessageInternal(Message<?> message) {
+		Assert.notNull(this.correlationStrategy, "'correlationStrategy' must not be null");
 		Object correlationKey = this.correlationStrategy.getCorrelationKey(message);
+		Assert.notNull(correlationKey, "The correlation key is required");
 		Object lock = getLock(correlationKey);
 		synchronized (lock) {
 			this.store.addMessagesToGroup(correlationKey, message);
@@ -103,12 +110,14 @@ private Object getLock(Object correlationKey) {
 
 	@SuppressWarnings("unchecked")
 	@Override
+	@Nullable
 	public Message<Object> receive() {
 		for (Object key : this.correlationLocks.keySet()) {
 			Object lock = getLock(key);
 			synchronized (lock) {
 				MessageGroup group = this.store.getMessageGroup(key);
 				//group might be removed by another thread
+				Assert.notNull(this.releaseStrategy, "'releaseStrategy' must not be null");
 				if (group != null && this.releaseStrategy.canRelease(group)) {
 					Message<?> nextMessage = null;
 

spring-integration-core/src/main/java/org/springframework/integration/aggregator/CorrelationStrategy.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,6 +16,8 @@
 
 package org.springframework.integration.aggregator;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.messaging.Message;
 
 /**
@@ -35,6 +37,7 @@ public interface CorrelationStrategy {
 	 * @param message The message.
 	 * @return The correlation key.
 	 */
+	@Nullable
 	Object getCorrelationKey(Message<?> message);
 
 }

spring-integration-core/src/main/java/org/springframework/integration/aggregator/DelegatingMessageGroupProcessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2020 the original author or authors.
+ * Copyright 2019-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -61,6 +61,7 @@ public class DelegatingMessageGroupProcessor implements MessageGroupProcessor, B
 
 	private volatile boolean messageBuilderFactorySet;
 
+	@SuppressWarnings("NullAway.Init")
 	private BeanFactory beanFactory;
 
 	public DelegatingMessageGroupProcessor(MessageGroupProcessor delegate,

spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2024 the original author or authors.
+ * Copyright 2002-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,6 +21,7 @@
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.integration.store.MessageGroup;
+import org.springframework.util.Assert;
 
 /**
  * A {@link MessageGroupProcessor} implementation that evaluates a SpEL expression. The SpEL context root is the list of
@@ -59,7 +60,9 @@ public void setExpectedType(Class<?> expectedType) {
 	 */
 	@Override
 	protected Object aggregatePayloads(MessageGroup group, Map<String, Object> headers) {
-		return this.processor.process(group.getMessages());
+		Object object = this.processor.process(group.getMessages());
+		Assert.state(object != null, "The process returned a null result. Null result is not expected.");
+		return object;
 	}
 
 }

spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageListProcessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2025 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,6 +18,8 @@
 
 import java.util.Collection;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.expression.Expression;
 import org.springframework.expression.ParseException;
 import org.springframework.integration.util.AbstractExpressionEvaluator;
@@ -37,6 +39,7 @@ public class ExpressionEvaluatingMessageListProcessor extends AbstractExpression
 
 	private final Expression expression;
 
+	@Nullable
 	private volatile Class<?> expectedType = null;
 
 	/**
@@ -102,7 +105,9 @@ public void setExpectedType(Class<?> expectedType) {
 	 */
 	@Override
 	public Object process(Collection<? extends Message<?>> messages) {
-		return this.evaluateExpression(this.expression, messages, this.expectedType);
+		Object object = this.evaluateExpression(this.expression, messages, this.expectedType);
+		Assert.state(object != null, "The evaluation of the expression returned a null. Null result is not expected." + this.expression);
+		return object;
 	}
 
 }