Fix nullability in the FluxAggregatorMessageHandler

* Make `windowSizeFunction` as `Function<Message<?>, @nullable Integer>`
because `sequenceSizeHeader()` may return `null` from message headers
* Extract local `subscriptionToDispose` in the `stop()` to satisfy null check context
* Use `Objects.requireNonNull(signal.get())` to satisfy `Function.apply()` contract.
The `if (signal.hasValue()) {` does the trick for us, but currently that is not visible
for that `signal.get()`
* Remove `@NullUnmarked` since we have just mitigated all the null problems

Author: artembilan

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

@@ -17,11 +17,11 @@
 package org.springframework.integration.aggregator;
 
 import java.time.Duration;
+import java.util.Objects;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 import java.util.function.Predicate;
 
-import org.jspecify.annotations.NullUnmarked;
 import org.jspecify.annotations.Nullable;
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
@@ -67,8 +67,8 @@ public class FluxAggregatorMessageHandler extends AbstractMessageProducingHandle
 	@Nullable
 	private Predicate<Message<?>> boundaryTrigger;
 
-	@Nullable
-	public Function<Message<?>, Integer> windowSizeFunction = FluxAggregatorMessageHandler::sequenceSizeHeader;
+	private Function<Message<?>, @Nullable Integer> windowSizeFunction =
+			FluxAggregatorMessageHandler::sequenceSizeHeader;
 
 	@Nullable
 	private Function<Flux<Message<?>>, Flux<Flux<Message<?>>>> windowConfigurer;
@@ -109,7 +109,6 @@ private Flux<Message<?>> releaseBy(Flux<Message<?>> groupFlux) {
 				.flatMap((windowFlux) -> windowFlux.transform(this.combineFunction));
 	}
 
-	@NullUnmarked
 	private Flux<Flux<Message<?>>> applyWindowOptions(Flux<Message<?>> groupFlux) {
 		if (this.boundaryTrigger != null) {
 			return groupFlux.windowUntil(this.boundaryTrigger);
@@ -118,7 +117,7 @@ private Flux<Flux<Message<?>>> applyWindowOptions(Flux<Message<?>> groupFlux) {
 				.switchOnFirst((signal, group) -> {
 					if (signal.hasValue()) {
 						Assert.notNull(this.windowSizeFunction, "'windowSizeFunction' must not be null");
-						Integer maxSize = this.windowSizeFunction.apply(signal.get());
+						Integer maxSize = this.windowSizeFunction.apply(Objects.requireNonNull(signal.get()));
 						if (maxSize != null) {
 							if (this.windowTimespan != null) {
 								return group.windowTimeout(maxSize, this.windowTimespan);
@@ -203,7 +202,7 @@ public void setWindowSize(int windowSize) {
 	 * @see Flux#window(int)
 	 * @see Flux#windowTimeout(int, Duration)
 	 */
-	public void setWindowSizeFunction(Function<Message<?>, Integer> windowSizeFunction) {
+	public void setWindowSizeFunction(Function<Message<?>, @Nullable Integer> windowSizeFunction) {
 		Assert.notNull(windowSizeFunction, "'windowSizeFunction' must not be null");
 		this.windowSizeFunction = windowSizeFunction;
 	}
@@ -255,8 +254,9 @@ public void start() {
 
 	@Override
 	public void stop() {
-		if (this.subscribed.compareAndSet(true, false) && this.subscription != null) {
-			this.subscription.dispose();
+		Disposable subscriptionToDispose = this.subscription;
+		if (this.subscribed.compareAndSet(true, false) && subscriptionToDispose != null) {
+			subscriptionToDispose.dispose();
 		}
 	}
 
@@ -289,8 +289,7 @@ private Mono<Message<?>> messageForWindowFlux(Flux<Message<?>> messageFlux) {
 								.build());
 	}
 
-	@NullUnmarked
-	private static Integer sequenceSizeHeader(Message<?> message) {
+	private static @Nullable Integer sequenceSizeHeader(Message<?> message) {
 		return message.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, Integer.class);
 	}