GH-9863: Deprecate PollerMetadata.sendTimeout option

Fixes: #9863
Issue link: #9863

The `PollerMetadata.sendTimeout` has been migrated to the `SourcePollingChannelAdapter.setSendTimeout()` long time ago.
Right now this option is not propagated anywhere

* Introduce missed `SourcePollingChannelAdapterSpec.sendTimeout`
* Deprecate (for removal) `PollerMetadata.sendTimeout`
* Verify `SourcePollingChannelAdapterSpec.sendTimeout` option in the `ReactiveStreamsTests`

**Auto-cherry-pick to `6.4.x` & `6.3.x`**

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/dsl/PollerSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2020 the original author or authors.
+ * Copyright 2016-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.
@@ -189,8 +189,14 @@ public PollerSpec taskExecutor(Executor taskExecutor) {
 		return this;
 	}
 
+	/**
+	 * The timeout for blocking send on channels.
+	 * @param sendTimeout the timeout to use.
+	 * @return the spec.
+	 * @deprecated in favor of {@link SourcePollingChannelAdapterSpec#sendTimeout(long)}
+	 */
+	@Deprecated(forRemoval = true, since = "6.3.9")
 	public PollerSpec sendTimeout(long sendTimeout) {
-		this.target.setSendTimeout(sendTimeout);
 		return this;
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2024 the original author or authors.
+ * Copyright 2016-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.
@@ -72,4 +72,15 @@ public SourcePollingChannelAdapterSpec taskScheduler(TaskScheduler taskScheduler
 		return this;
 	}
 
+	/**
+	 * The timeout for blocking send on channels.
+	 * @param sendTimeout the timeout to use.
+	 * @return the spec.
+	 * @since 6.3.9
+	 */
+	public SourcePollingChannelAdapterSpec sendTimeout(long sendTimeout) {
+		this.endpointFactoryBean.setSendTimeout(sendTimeout);
+		return this;
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.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.
@@ -138,10 +138,12 @@ public Executor getTaskExecutor() {
 		return this.taskExecutor;
 	}
 
+	@Deprecated(forRemoval = true, since = "6.3.9")
 	public long getSendTimeout() {
 		return this.sendTimeout;
 	}
 
+	@Deprecated(forRemoval = true, since = "6.3.9")
 	public void setSendTimeout(long sendTimeout) {
 		this.sendTimeout = sendTimeout;
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2024 the original author or authors.
+ * Copyright 2016-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.
@@ -52,6 +52,7 @@
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.endpoint.ReactiveMessageSourceProducer;
 import org.springframework.integration.endpoint.ReactiveStreamsConsumer;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.support.GenericMessage;
@@ -122,6 +123,7 @@ void testReactiveFlow() throws Exception {
 
 		disposable.dispose();
 		assertThat(this.messageSource.isRunning()).isFalse();
+		assertThat(TestUtils.getPropertyValue(this.messageSource, "messagingTemplate.sendTimeout")).isEqualTo(256L);
 	}
 
 	@Test
@@ -275,6 +277,7 @@ public Publisher<Message<String>> reactiveFlow() {
 			return IntegrationFlow
 					.from(() -> new GenericMessage<>("a,b,c,d,e,f"),
 							e -> e.poller(p -> p.trigger(ctx -> this.invoked.getAndSet(true) ? null : Instant.now()))
+									.sendTimeout(256)
 									.id("reactiveStreamsMessageSource"))
 					.split(String.class, p -> p.split(","))
 					.log()