GH-10547: DefaultHeaderChannelRegistry: Fix race condition

Fixes: #10547

When stopping an application, the following error log is generated:
```
ERROR Unexpected error occurred in scheduled task org.springframework.core.task.TaskRejectedException: ExecutorService in shutdown state did not accept task: bean 'integrationHeaderChannelRegistry'
 at org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler.schedule(ThreadPoolTaskScheduler.java:430)
Caused by: java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask@7bc9f638[Not completed, task = java.util.concurrent.Executors$RunnableAdapter@3ecb6f85[Wrapped task = DelegatingErrorHandlingRunnable for bean 'integrationHeaderChannelRegistry']] rejected from org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler$1@57715118[Shutting down, pool size = 1, active threads = 1, queued tasks = 0, completed tasks = 1225]
```

* Check for `isRunning()` in the `DefaultHeaderChannelRegistry.run()` before scheduling a new reaper task
* Increase latch wait timeout to 20 seconds in the `SftpSessionFactoryTests.sharedSessionConcurrentAccess()`

(cherry picked from commit 5160117)

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultHeaderChannelRegistry.java

@@ -56,11 +56,11 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport
 
 	private static final int DEFAULT_REAPER_DELAY = 60000;
 
-	protected static final AtomicLong id = new AtomicLong(); // NOSONAR
+	protected static final AtomicLong id = new AtomicLong();
 
-	protected final Map<String, MessageChannelWrapper> channels = new ConcurrentHashMap<>(); // NOSONAR
+	protected final Map<String, MessageChannelWrapper> channels = new ConcurrentHashMap<>();
 
-	protected final String uuid = UUID.randomUUID() + ":"; // NOSONAR
+	protected final String uuid = UUID.randomUUID() + ":";
 
 	private boolean removeOnGet;
 
@@ -118,20 +118,13 @@ public final int size() {
 		return this.channels.size();
 	}
 
-	@Override
-	protected void onInit() {
-		super.onInit();
-		Assert.notNull(getTaskScheduler(), "a task scheduler is required");
-	}
-
 	@Override
 	public void start() {
 		this.lock.lock();
 		try {
 			if (!this.running) {
-				Assert.notNull(getTaskScheduler(), "a task scheduler is required");
-				this.reaperScheduledFuture = getTaskScheduler()
-						.schedule(this, Instant.now().plusMillis(this.reaperDelay));
+				this.reaperScheduledFuture =
+						getTaskScheduler().schedule(this, Instant.now().plusMillis(this.reaperDelay));
 
 				this.running = true;
 			}
@@ -177,13 +170,12 @@ public Object channelToChannelName(@Nullable Object channel) {
 	@Override
 	@Nullable
 	public Object channelToChannelName(@Nullable Object channel, long timeToLive) {
-		if (!this.running && !this.explicitlyStopped && this.getTaskScheduler() != null) {
+		if (!this.running && !this.explicitlyStopped) {
 			start();
 		}
-		if (channel instanceof MessageChannel) {
+		if (channel instanceof MessageChannel messageChannel) {
 			String name = this.uuid + id.incrementAndGet();
-			this.channels.put(name, new MessageChannelWrapper((MessageChannel) channel,
-					System.currentTimeMillis() + timeToLive));
+			this.channels.put(name, new MessageChannelWrapper(messageChannel, System.currentTimeMillis() + timeToLive));
 			logger.debug(() -> "Registered " + channel + " as " + name);
 			return name;
 		}
@@ -246,8 +238,10 @@ public void run() {
 					iterator.remove();
 				}
 			}
-			this.reaperScheduledFuture = getTaskScheduler()
-					.schedule(this, Instant.now().plusMillis(this.reaperDelay));
+			if (isRunning()) {
+				this.reaperScheduledFuture =
+						getTaskScheduler().schedule(this, Instant.now().plusMillis(this.reaperDelay));
+			}
 
 			logger.trace(() -> "Reaper completed; channels size=" + this.channels.size());
 		}

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java

@@ -331,15 +331,15 @@ protected SftpClient createSftpClient(ClientSession clientSession,
 				});
 			}
 
-			assertThat(executionLatch.await(10, TimeUnit.SECONDS)).isTrue();
+			assertThat(executionLatch.await(20, TimeUnit.SECONDS)).isTrue();
 			synchronized (errors) {
 				assertThat(errors).isEmpty();
 			}
 
 			assertThat(clientInstances).hasValue(1);
 
 			executorService.shutdown();
-			assertThat(executorService.awaitTermination(10, TimeUnit.SECONDS)).isTrue();
+			assertThat(executorService.awaitTermination(20, TimeUnit.SECONDS)).isTrue();
 
 			sftpSessionFactory.destroy();
 		}