Deadlock when using SFTP CachingSessionFactory with FileExistsMode.FAIL

[javierestevez]

Affects 5.2.5

It looks like there is a deadlock when using a CachingSessionFactory and FileExistsMode.FAIL. When FileExistsMode.FAIL is used, then we first check if the file exists in the server by calling RemoteFileTemplate#exists which in turn calls RemoteFileTemplate#execute. The latter requests a new session to the session factory instead of using the session already obtained (RemoteFileTemplate#activeTemplateCallbacks size always evaluated to 0 during my tests). This can end up in a deadlock situation if there are as many (or less) cached sessions as the number of concurrent threads (possibly with more too).

Below is a test I used to reproduce this locally using 2 concurrent threads and 2 cached sessions. I omitted some details for brevity. Hopefully it is enough to reproduce it on your side.

private final Runnable runnable = () -> {
    UUID uuid = UUID.randomUUID();
    String filename = uuid + ".tst";
    LOGGER.info("Uploading file " + filename + " to SFTP...");

    testSftpGateway.send(DIRECTORY_NAME, filename, "Some bytes".getBytes());
    LOGGER.info("... Uploaded " + filename + "!");
};

@Test
public void test() {
    withConfiguredSftpServer(sftpServer -> {
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        for (int i = 0; i < 100; i++) {
            executorService.execute(runnable);
        }
        executorService.shutdown();
        executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
        LOGGER.info("Finished all threads");
    });
}
@MessagingGateway
interface TestSftpGateway {
    @Gateway(requestChannel = OUTBOUND_CHANEL)
    void send(@Header(FileHeaders.REMOTE_DIRECTORY) String directoryName,
              @Header(FileHeaders.FILENAME) String fileName,
              @Payload byte[] payload);
}

@Configuration
@IntegrationComponentScan
@EnableIntegration
static class TestConfiguration {

    @Bean
    public SessionFactory<ChannelSftp.LsEntry> sftpSessionFactory() {
        DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
        factory.setHost("localhost");
        factory.setPort(22);
        factory.setUser("foo");
        factory.setPassword("pass");
        factory.setAllowUnknownKeys(true);

        CachingSessionFactory<ChannelSftp.LsEntry> cachingSessionFactory = new CachingSessionFactory<>(factory, 2);
        cachingSessionFactory.setTestSession(true);
        cachingSessionFactory.setSessionWaitTimeout(1000);

        return cachingSessionFactory;
    }

    @Bean
    public SftpRemoteFileTemplate sftpRemoteFileTemplate(SessionFactory<ChannelSftp.LsEntry> sftpSessionFactory) {
        FunctionExpression<Message<?>> remoteDirectoryExpression = new FunctionExpression<>(this::getRemoteDirectoryHeaderValue);
        SftpRemoteFileTemplate sftpRemoteFileTemplate = new SftpRemoteFileTemplate(sftpSessionFactory);
        sftpRemoteFileTemplate.setRemoteDirectoryExpression(remoteDirectoryExpression);
        return sftpRemoteFileTemplate;
    }

    @Bean
    public IntegrationFlow outboundFlow(SftpRemoteFileTemplate template) {
        return IntegrationFlows.from(OUTBOUND_CHANEL)
                .handle(Sftp.outboundAdapter(template, FileExistsMode.FAIL)
                        .useTemporaryFileName(false)
                        .remoteDirectory(this::getRemoteDirectoryHeaderValue))
                .get();
    }

    private String getRemoteDirectoryHeaderValue(Message<?> message) {
        MessageHeaders messageHeaders = message.getHeaders();
        return (String) messageHeaders.get(FileHeaders.REMOTE_DIRECTORY);
    }
}

[artembilan]

The RemoteFileTemplate.doSend() has to be fixed:

	private void doSend(Session<F> session, FileExistsMode mode, String remoteFilePath, String tempFilePath,
			InputStream stream) throws IOException {

	...
		else {
			if (exists(remoteFilePath)) {
		...
			}
		}
	}

We should not call that exists(remoteFilePath) on the template, but rather a provided session.exists(remoteFilePath).

Good catch!

I'll try to fix it over weekend.
Thank you for the code to let us to reproduce!

[artembilan]

Reopen: merged to the master accidentally: ... 😢

See linked PR for 5.2.x branch.

[artembilan]

Fixed with the merged PR.