Spring Integration遇SftpException后SSH连接泄漏致SFTP过载

版本信息 #

• spring-boot-starter-parent：3.2.3
• Java：17

Maven依赖 #

• spring-boot-starter-integration
• spring-integration-sftp

问题描述 #

使用Spring Integration对接SFTP服务器时，无异常场景下会话可正常复用；但触发SftpException（如文件不存在、权限不足等）后，SSH连接数会随每次错误持续增长，最终导致SFTP服务器过载（曾出现超300个连接）。

核心代码 #

@Bean
public CachingSessionFactory<SftpClient.DirEntry> sftpSessionFactory(
        final SftpPrivateKeyConfigurationProperties privateKeyConfigurationProperties,
        final SftpConfigurationProperties sftpConfigurationProperties) {
    final DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
    factory.setHost(privateKeyConfigurationProperties.getHost());
    factory.setPort(privateKeyConfigurationProperties.getPort());
    factory.setUser(privateKeyConfigurationProperties.getUser());
    factory.setPrivateKey(new ByteArrayResource(privateKeyConfigurationProperties.getPrivateKey().getBytes()));
    factory.setAllowUnknownKeys(true);
    factory.setTimeout(Math.toIntExact(sftpConfigurationProperties.getTimeout().toMillis()));
    var cachedFactory = new CachingSessionFactory<>(factory, sftpConfigurationProperties.getPoolSize());
    cachedFactory.setSessionWaitTimeout(sftpConfigurationProperties.getSessionWaitTimeout().toMillis());
    return cachedFactory;
}

@Bean
IntegrationFlow flowSubAccount(final SftpConfigurationProperties sftpConfigurationProperties,
        final CachingSessionFactory<SftpClient.DirEntry> sessionFactory,
        @Qualifier("sftpFilter") final ChainFileListFilter<SftpClient.DirEntry> sftpFilter,
        final ConfigurableApplicationContext ctx)
        throws IOException {

    return IntegrationFlow.from(
            Sftp.inboundAdapter(sessionFactory, Comparator.comparingLong(File::lastModified))
                    .preserveTimestamp(true)
                    .deleteRemoteFiles(sftpConfigurationProperties.getDeleteRemoteFiles())
                    .remoteDirectory(sftpConfigurationProperties.getRemoteDirectory())
                    .filter(sftpFilter)
                    .localDirectory(new File(sftpConfigurationProperties.getLocalDirectory()))
                    .localFilenameExpression("#root")
                    .autoCreateLocalDirectory(true),
            e -> e.autoStartup(true).poller(Pollers.fixedDelay(sftpConfigurationProperties.getPollDuration()).maxMessagesPerPoll(1).errorHandler((ex) -> {
                log.error("An error has been encountered", ex);
            })))
            .route(Message.class, this::payloadTypeSubAccountRouter)
            .get();
}

异常堆栈 #

org.springframework.messaging.MessagingException: Problem occurred while synchronizing 'this/path/does/not/exists' to local directory
    at org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer.synchronizeToLocalDirectory(AbstractInboundFileSynchronizer.java:348) ~[spring-integration-file-6.2.2.jar:6.2.2]
    at org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizingMessageSource.doReceive(AbstractInboundFileSynchronizingMessageSource.java:267) ~[spring-integration-file-6.2.2.jar:6.2.2]
    at org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizingMessageSource.doReceive(AbstractInboundFileSynchronizingMessageSource.java:69) ~[spring-integration-file-6.2.2.jar:6.2.2]
    at org.springframework.integration.endpoint.AbstractFetchLimitingMessageSource.doReceive(AbstractFetchLimitingMessageSource.java:47) ~[spring-integration-core-6.2.2.jar:6.2.2]
    at org.springframework.integration.endpoint.AbstractMessageSource.receive(AbstractMessageSource.java:142) ~[spring-integration-core-6.2.2.jar:6.2.2]
    at org.springframework.integration.endpoint.SourcePollingChannelAdapter.receiveMessage(SourcePollingChannelAdapter.java:222) ~[spring-integration-core-6.2.2.jar:6.2.2]
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:450) ~[spring-integration-core-6.2.2.jar:6.2.2]
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.pollForMessage(AbstractPollingEndpoint.java:419) ~[spring-integration-core-6.2.2.jar:6.2.2]
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$4(AbstractPollingEndpoint.java:355) ~[spring-integration-core-6.2.2.jar:6.2.2]
    at org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute$0(ErrorHandlingTaskExecutor.java:57) ~[spring-integration-core-6.2.2.jar:6.2.2]
    at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50) ~[spring-core-6.1.4.jar:6.1.4]
    at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:55) ~[spring-integration-core-6.2.2.jar:6.2.2]
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$5(AbstractPollingEndpoint.java:348) ~[spring-integration-core-6.2.2.jar:6.2.2]
    at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54) ~[spring-context-6.1.4.jar:6.1.4]
    at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:96) ~[spring-context-6.1.4.jar:6.1.4]
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) ~[na:na]
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[na:na]
    at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) ~[na:na]
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[na:na]
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[na:na]
    at java.base/java.lang.Thread.run(Thread.java:833) ~[na:na]
Caused by: org.springframework.messaging.MessagingException: Failed to execute on session
    at org.springframework.integration.file.remote.RemoteFileTemplate.execute(RemoteFileTemplate.java:461) ~[spring-integration-file-6.2.2.jar:6.2.2]
    at org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer.synchronizeToLocalDirectory(AbstractInboundFileSynchronizer.java:341) ~[spring-integration-file-6.2.2.jar:6.2.2]
    ... 20 common frames omitted
Caused by: org.apache.sshd.sftp.common.SftpException: No such file
    at org.apache.sshd.sftp.client.impl.AbstractSftpClient.throwStatusException(AbstractSftpClient.java:277) ~[sshd-sftp-2.11.0.jar:2.11.0]
    at org.apache.sshd.sftp.client.impl.AbstractSftpClient.checkAttributesResponse(AbstractSftpClient.java:333) ~[sshd-sftp-2.11.0.jar:2.11.0]
    at org.apache.sshd.sftp.client.impl.AbstractSftpClient.checkAttributes(AbstractSftpClient.java:325) ~[sshd-sftp-2.11.0.jar:2.11.0]
    at org.apache.sshd.sftp.client.impl.AbstractSftpClient.lstat(AbstractSftpClient.java:1010) ~[sshd-sftp-2.11.0.jar:2.11.0]
    at org.springframework.integration.sftp.session.SftpSession.doList(SftpSession.java:102) ~[spring-integration-sftp-6.2.2.jar:6.2.2]
    at org.springframework.integration.sftp.session.SftpSession.list(SftpSession.java:80) ~[spring-integration-sftp-6.2.2.jar:6.2.2]
    at org.springframework.integration.sftp.session.SftpSession.list(SftpSession.java:52) ~[spring-integration-sftp-6.2.2.jar:6.2.2]
    at org.springframework.integration.file.remote.session.CachingSessionFactory$CachedSession.list(CachingSessionFactory.java:246) ~[spring-integration-file-6.2.2.jar:6.2.2]
    at org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer.transferFilesFromRemoteToLocal(AbstractInboundFileSynchronizer.java:356) ~[spring-integration-file-6.2.2.jar:6.2.2]
    at org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer.lambda$synchronizeToLocalDirectory$0(AbstractInboundFileSynchronizer.java:342) ~[spring-integration-file-6.2.2.jar:6.2.2]
    at org.springframework.integration.file.remote.RemoteFileTemplate.execute(RemoteFileTemplate.java:452) ~[spring-integration-file-6.2.2.jar:6.2.2]
    ... 21 common frames omitted

排查与尝试 #

• 替换CachingSessionFactory为DefaultSftpSessionFactory，问题更严重；
• 开启Spring Integration TRACE日志，发现异常时会话被标记为脏连接并从连接池移除，但该连接仍处于活跃状态未关闭；
• 尝试关闭整个连接池重建，引发其他问题；
• 考虑捕获异常时停止并重启流，但方案不合理。

求助 #

如何彻底关闭这些被标记为脏连接的活跃SSH连接？

内容的提问来源于stack exchange，提问作者Nicolas