Add nullability to org.springframework.aggregator & org.springframework.acks packages

Note in this commit that there cases where methods are called where the parameters passed in may contain a null.
In those cases the attribute that contains the potential null is tested with an `Assert.notNull`.  This may not be the best approach, let's discuss.

Also note that in FluxAggregatorMessageHandler line 112 `@NullUnmarked` is used on the the applyWindowOptions method.
This is because the consumer.apply on line 121 can not be null according to JSpecify.   But we can't make that assumption for how the user implemented the function.
There maybe a better way of handling this besides marking the method as `NullUnmarked`.   Let's discuss.

Similarly in FluxAggregatorMessageHandler the way in which sequenceSizeHeader is utilized it had to be `@NullUnmarked`.   Let's discuss

Make sure all annotations are applied consistently

Author: cppwfs

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.
@@ -21,6 +21,7 @@
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
@@ -56,6 +57,7 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag
 
 	private boolean messageBuilderFactorySet;
 
+	@SuppressWarnings("NullAway.Init")
 	private BeanFactory beanFactory;
 
 	@Override
@@ -117,6 +119,7 @@ protected Map<String, Object> aggregateHeaders(MessageGroup group) {
 		return getHeadersFunction().apply(group);
 	}
 
+	@Nullable
 	protected abstract Object aggregatePayloads(MessageGroup group, Map<String, Object> defaultHeaders);
 
 }

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

@@ -65,6 +65,7 @@
 import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.core.DestinationResolutionException;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.scheduling.TaskScheduler;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.ObjectUtils;
@@ -125,8 +126,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 +146,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 +173,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 +513,7 @@ public MessageChannel getDiscardChannel() {
 		return this.discardChannel;
 	}
 
+	@Nullable
 	protected String getDiscardChannelName() {
 		return this.discardChannelName;
 	}
@@ -532,6 +542,7 @@ protected boolean isReleasePartialSequences() {
 		return this.releasePartialSequences;
 	}
 
+	@Nullable
 	protected Expression getGroupTimeoutExpression() {
 		return this.groupTimeoutExpression;
 	}
@@ -641,8 +652,10 @@ protected boolean isExpireGroupsUponCompletion() {
 	}
 
 	private void removeEmptyGroupAfterTimeout(UUID groupId, long timeout) {
+		TaskScheduler taskScheduler = getTaskScheduler();
+		Assert.notNull(taskScheduler, "'taskScheduler' must not be null");
 		ScheduledFuture<?> scheduledFuture =
-				getTaskScheduler()
+				taskScheduler
 						.schedule(() -> {
 							Lock lock = this.lockRegistry.obtain(groupId.toString());
 
@@ -699,8 +712,10 @@ else if ((Long) groupTimeout > 0) {
 				Object groupId = messageGroup.getGroupId();
 				long timestamp = messageGroup.getTimestamp();
 				long lastModified = messageGroup.getLastModified();
+				TaskScheduler taskScheduler = getTaskScheduler();
+				Assert.notNull(taskScheduler, "'taskScheduler' must not be null");
 				ScheduledFuture<?> scheduledFuture =
-						getTaskScheduler()
+						taskScheduler
 								.schedule(() -> {
 									try {
 										processForceRelease(groupId, timestamp, lastModified);
@@ -753,7 +768,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 +927,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 +942,7 @@ protected Collection<Message<?>> completeGroup(Message<?> message, Object correl
 			this.logger.debug(() -> "Completing group with correlationKey [" + correlationKey + "]");
 
 			result = this.outputProcessor.processMessageGroup(group);
+			Assert.notNull(result, "the group returned a null result");
 			if (isResultCollectionOfMessages(result)) {
 				partialSequence = (Collection<Message<?>>) result;
 			}
@@ -988,6 +1002,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);
@@ -1024,7 +1039,9 @@ public void start() {
 			if (this.expireTimeout > 0) {
 				purgeOrphanedGroups();
 				if (this.expireDuration != null) {
-					getTaskScheduler()
+					TaskScheduler taskScheduler = getTaskScheduler();
+					Assert.notNull(taskScheduler, "'taskScheduler' must not be null");
+					taskScheduler
 							.scheduleWithFixedDelay(this::purgeOrphanedGroups, this.expireDuration);
 				}
 			}
@@ -1062,6 +1079,7 @@ public void purgeOrphanedGroups() {
 
 	protected static class SequenceAwareMessageGroup extends SimpleMessageGroup {
 
+		@Nullable
 		private final SimpleMessageGroup sourceGroup;
 
 		public SequenceAwareMessageGroup(MessageGroup messageGroup) {
@@ -1124,6 +1142,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.
@@ -18,6 +18,8 @@
 
 import java.util.Map;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.integration.store.MessageGroup;
@@ -58,6 +60,7 @@ public void setExpectedType(Class<?> expectedType) {
 	 * {@link org.springframework.integration.core.MessagingTemplate} to send downstream.
 	 */
 	@Override
+	@Nullable
 	protected Object aggregatePayloads(MessageGroup group, Map<String, Object> headers) {
 		return this.processor.process(group.getMessages());
 	}