Fix nullability in the FluxAggregatorMessageHandler

* This PR is not ready for merge.   There are failing tests, but wanted to capture where it is at.
* Also still having issue with the .apply() in 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

Updated tests so that they work with nullability changes

Update the tests so that they will pass with nullify changes

Author: artembilan

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

@@ -17,7 +17,6 @@
 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;
@@ -203,7 +202,7 @@ public void setWindowSize(int windowSize) {
 	 * @see Flux#window(int)
 	 * @see Flux#windowTimeout(int, Duration)
 	 */
-	public void setWindowSizeFunction(Function<Message<?>, @Nullable  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,9 +289,7 @@ private Mono<Message<?>> messageForWindowFlux(Flux<Message<?>> messageFlux) {
 								.build());
 	}
 
-
-	private static Integer sequenceSizeHeader(Message<?> message) {
-		return Objects.requireNonNull(message.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, Integer.class));
+	private static @Nullable Integer sequenceSizeHeader(Message<?> message) {
+		return message.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, Integer.class);
 	}
-
 }

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

@@ -18,8 +18,6 @@
 
 import java.util.Collection;
 
-import org.jspecify.annotations.Nullable;
-
 import org.springframework.messaging.Message;
 
 /**

spring-integration-core/src/test/java/org/springframework/integration/aggregator/FluxAggregatorMessageHandlerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2024 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.
@@ -29,13 +29,19 @@
 import reactor.core.publisher.Flux;
 import reactor.test.StepVerifier;
 
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.scheduling.TaskScheduler;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 /**
  * @author Artem Bilan
@@ -159,6 +165,7 @@ void testCustomCombineFunction() {
 	void testWindowTimespan() {
 		QueueChannel resultChannel = new QueueChannel();
 		FluxAggregatorMessageHandler fluxAggregatorMessageHandler = new FluxAggregatorMessageHandler();
+		fluxAggregatorMessageHandler.setBeanFactory(getBeanFactory());
 		fluxAggregatorMessageHandler.setOutputChannel(resultChannel);
 		fluxAggregatorMessageHandler.setWindowTimespan(Duration.ofMillis(100));
 		fluxAggregatorMessageHandler.start();
@@ -213,6 +220,15 @@ void testWindowTimespan() {
 		executorService.shutdown();
 	}
 
+	private BeanFactory getBeanFactory() {
+		BeanFactory beanFactory = mock(BeanFactory.class);
+		TaskScheduler taskScheduler = mock(TaskScheduler.class);
+		when(beanFactory.getBean(eq("taskScheduler"), any(Class.class)))
+				.thenReturn(taskScheduler);
+		when(beanFactory.containsBean("taskScheduler")).thenReturn(true);
+		return beanFactory;
+	}
+
 	@Test
 	void testBoundaryTrigger() {
 		QueueChannel resultChannel = new QueueChannel();

spring-integration-core/src/test/java/org/springframework/integration/channel/registry/HeaderChannelRegistryTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2024 the original author or authors.
+ * Copyright 2013-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.
@@ -46,6 +46,8 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -238,6 +240,7 @@ public void testBFCRNoRegistry() {
 	@Test
 	public void testRemoveOnGet() {
 		DefaultHeaderChannelRegistry registry = new DefaultHeaderChannelRegistry();
+		registry.setBeanFactory(getBeanFactory());
 		MessageChannel channel = new DirectChannel();
 		String foo = (String) registry.channelToChannelName(channel);
 		Map<?, ?> map = TestUtils.getPropertyValue(registry, "channels", Map.class);
@@ -249,6 +252,15 @@ public void testRemoveOnGet() {
 		assertThat(map.size()).isEqualTo(0);
 	}
 
+	private BeanFactory getBeanFactory() {
+		BeanFactory beanFactory = mock(BeanFactory.class);
+		TaskScheduler taskScheduler = mock(TaskScheduler.class);
+		when(beanFactory.getBean(eq("taskScheduler"), any(Class.class)))
+				.thenReturn(taskScheduler);
+		when(beanFactory.containsBean("taskScheduler")).thenReturn(true);
+		return beanFactory;
+	}
+
 	public static class Foo extends AbstractReplyProducingMessageHandler {
 
 		@Override

spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java

@@ -36,6 +36,7 @@
 import reactor.core.scheduler.Schedulers;
 import reactor.test.StepVerifier;
 
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.ConfigurableApplicationContext;
@@ -57,10 +58,15 @@
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.scheduling.TaskScheduler;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 /**
  * @author Artem Bilan
@@ -130,6 +136,7 @@ void testReactiveFlow() throws Exception {
 	@Test
 	void testPollableReactiveFlow() throws Exception {
 		assertThat(this.reactiveTransformer).isInstanceOf(ReactiveStreamsConsumer.class);
+		this.reactiveTransformer.setBeanFactory(getBeanFactory());
 		this.inputChannel.send(new GenericMessage<>("1,2,3,4,5"));
 
 		CountDownLatch latch = new CountDownLatch(6);
@@ -165,6 +172,15 @@ void testPollableReactiveFlow() throws Exception {
 		exec.shutdownNow();
 	}
 
+	private BeanFactory getBeanFactory() {
+		BeanFactory beanFactory = mock(BeanFactory.class);
+		TaskScheduler taskScheduler = mock(TaskScheduler.class);
+		when(beanFactory.getBean(eq("taskScheduler"), any(Class.class)))
+				.thenReturn(taskScheduler);
+		when(beanFactory.containsBean("taskScheduler")).thenReturn(true);
+		return beanFactory;
+	}
+
 	@Test
 	void testFromPublisher() {
 		Flux<Message<?>> messageFlux =

spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.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.
@@ -42,11 +42,15 @@
 import org.springframework.messaging.support.ChannelInterceptor;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.scheduling.TaskScheduler;
 import org.springframework.util.ReflectionUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 /**
  * @author Mark Fisher
@@ -65,7 +69,7 @@ public void futureWithMessageReturned() throws Exception {
 		GatewayProxyFactoryBean<TestEchoService> proxyFactory = new GatewayProxyFactoryBean<>(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
 		proxyFactory.setBeanName("testGateway");
-		proxyFactory.setBeanFactory(mock(BeanFactory.class));
+		proxyFactory.setBeanFactory(getBeanFactory());
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = proxyFactory.getObject();
 		Future<Message<?>> f = service.returnMessage("foo");
@@ -88,7 +92,7 @@ protected boolean doSend(Message<?> message, long timeout) {
 		GatewayProxyFactoryBean<TestEchoService> proxyFactory = new GatewayProxyFactoryBean<>(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(channel);
 		proxyFactory.setBeanName("testGateway");
-		proxyFactory.setBeanFactory(mock(BeanFactory.class));
+		proxyFactory.setBeanFactory(getBeanFactory());
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = proxyFactory.getObject();
 		Future<Message<?>> f = service.returnMessage("foo");
@@ -106,7 +110,7 @@ public void listenableFutureWithMessageReturned() throws Exception {
 		GatewayProxyFactoryBean<TestEchoService> proxyFactory = new GatewayProxyFactoryBean<>(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
 		proxyFactory.setBeanName("testGateway");
-		proxyFactory.setBeanFactory(mock(BeanFactory.class));
+		proxyFactory.setBeanFactory(getBeanFactory());
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = proxyFactory.getObject();
 		CompletableFuture<Message<?>> f = service.returnMessageListenable("foo");
@@ -132,7 +136,7 @@ public void customFutureReturned() {
 		GatewayProxyFactoryBean<TestEchoService> proxyFactory = new GatewayProxyFactoryBean<>(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
 		proxyFactory.setBeanName("testGateway");
-		proxyFactory.setBeanFactory(mock(BeanFactory.class));
+		proxyFactory.setBeanFactory(getBeanFactory());
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = proxyFactory.getObject();
 		CustomFuture f = service.returnCustomFuture("foo");
@@ -149,7 +153,7 @@ public void nonAsyncFutureReturned() {
 		GatewayProxyFactoryBean<TestEchoService> proxyFactory = new GatewayProxyFactoryBean<>(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
 		proxyFactory.setBeanName("testGateway");
-		proxyFactory.setBeanFactory(mock(BeanFactory.class));
+		proxyFactory.setBeanFactory(getBeanFactory());
 
 		proxyFactory.setAsyncExecutor(null);    // Not async - user flow returns Future<?>
 
@@ -181,7 +185,7 @@ public void futureWithPayloadReturned() throws Exception {
 		GatewayProxyFactoryBean<TestEchoService> proxyFactory = new GatewayProxyFactoryBean<>(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
 		proxyFactory.setBeanName("testGateway");
-		proxyFactory.setBeanFactory(mock(BeanFactory.class));
+		proxyFactory.setBeanFactory(getBeanFactory());
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = proxyFactory.getObject();
 		Future<String> f = service.returnString("foo");
@@ -197,7 +201,7 @@ public void futureWithWildcardReturned() throws Exception {
 		GatewayProxyFactoryBean<TestEchoService> proxyFactory = new GatewayProxyFactoryBean<>(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
 		proxyFactory.setBeanName("testGateway");
-		proxyFactory.setBeanFactory(mock(BeanFactory.class));
+		proxyFactory.setBeanFactory(getBeanFactory());
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = proxyFactory.getObject();
 		Future<?> f = service.returnSomething("foo");
@@ -211,7 +215,7 @@ public void futureVoid() throws Exception {
 		GatewayProxyFactoryBean<TestEchoService> proxyFactory = new GatewayProxyFactoryBean<>(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(new NullChannel());
 		proxyFactory.setBeanName("testGateway");
-		proxyFactory.setBeanFactory(mock(BeanFactory.class));
+		proxyFactory.setBeanFactory(getBeanFactory());
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = proxyFactory.getObject();
 		Future<Void> f = service.asyncSendAndForget("test1");
@@ -251,7 +255,7 @@ public void futureVoidReply() throws Exception {
 		GatewayProxyFactoryBean<TestEchoService> proxyFactory = new GatewayProxyFactoryBean<>(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
 		proxyFactory.setBeanName("testGateway");
-		proxyFactory.setBeanFactory(mock(BeanFactory.class));
+		proxyFactory.setBeanFactory(getBeanFactory());
 		proxyFactory.setAsyncExecutor(null);
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = proxyFactory.getObject();
@@ -267,7 +271,7 @@ public void monoWithMessageReturned() {
 		startResponder(requestChannel);
 		GatewayProxyFactoryBean<TestEchoService> proxyFactory = new GatewayProxyFactoryBean<>(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
-		proxyFactory.setBeanFactory(mock(BeanFactory.class));
+		proxyFactory.setBeanFactory(getBeanFactory());
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = proxyFactory.getObject();
@@ -282,7 +286,7 @@ public void monoWithPayloadReturned() {
 		startResponder(requestChannel);
 		GatewayProxyFactoryBean<TestEchoService> proxyFactory = new GatewayProxyFactoryBean<>(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
-		proxyFactory.setBeanFactory(mock(BeanFactory.class));
+		proxyFactory.setBeanFactory(getBeanFactory());
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = proxyFactory.getObject();
@@ -297,7 +301,7 @@ public void monoWithWildcardReturned() {
 		startResponder(requestChannel);
 		GatewayProxyFactoryBean<TestEchoService> proxyFactory = new GatewayProxyFactoryBean<>(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
-		proxyFactory.setBeanFactory(mock(BeanFactory.class));
+		proxyFactory.setBeanFactory(getBeanFactory());
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = proxyFactory.getObject();
@@ -313,7 +317,7 @@ public void monoWithConsumer() {
 		startResponder(requestChannel);
 		GatewayProxyFactoryBean<TestEchoService> proxyFactory = new GatewayProxyFactoryBean<>(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(requestChannel);
-		proxyFactory.setBeanFactory(mock(BeanFactory.class));
+		proxyFactory.setBeanFactory(getBeanFactory());
 		proxyFactory.setBeanName("testGateway");
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = proxyFactory.getObject();
@@ -329,7 +333,7 @@ public void monoVoid() throws InterruptedException {
 		GatewayProxyFactoryBean<TestEchoService> proxyFactory = new GatewayProxyFactoryBean<>(TestEchoService.class);
 		proxyFactory.setDefaultRequestChannel(new NullChannel());
 		proxyFactory.setBeanName("testGateway");
-		proxyFactory.setBeanFactory(mock(BeanFactory.class));
+		proxyFactory.setBeanFactory(getBeanFactory());
 		proxyFactory.afterPropertiesSet();
 		TestEchoService service = proxyFactory.getObject();
 		Mono<Void> mono = service.monoVoid("test1");
@@ -372,6 +376,15 @@ private static void startResponder(final PollableChannel requestChannel) {
 		}).start();
 	}
 
+	private BeanFactory getBeanFactory() {
+		BeanFactory beanFactory = mock(BeanFactory.class);
+		TaskScheduler taskScheduler = mock(TaskScheduler.class);
+		when(beanFactory.getBean(eq("taskScheduler"), any(Class.class)))
+				.thenReturn(taskScheduler);
+		when(beanFactory.containsBean("taskScheduler")).thenReturn(true);
+		return beanFactory;
+	}
+
 	private interface TestEchoService {
 
 		Future<String> returnString(String s);

spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.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.
@@ -89,6 +89,7 @@
 import org.springframework.messaging.handler.annotation.Payload;
 import org.springframework.messaging.support.ChannelInterceptor;
 import org.springframework.messaging.support.MessageHeaderAccessor;
+import org.springframework.scheduling.TaskScheduler;
 import org.springframework.stereotype.Component;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ActiveProfiles;
@@ -340,6 +341,7 @@ public void testWithServiceEquals() {
 		bf.registerSingleton("requestChannelBar", channel);
 		bf.registerSingleton("requestChannelBaz", channel);
 		bf.registerSingleton("requestChannelFoo", channel);
+		bf.registerSingleton("taskScheduler", mock(TaskScheduler.class));
 		fb.setBeanFactory(bf);
 		fb.afterPropertiesSet();
 		assertThat(fb.getObject()).isNotSameAs(bar);