`AbstractRemoteFileOutboundGateway`: concurrent GET into the same new local directory fails with "Failed to make local directory"

[dlwldn30]

In what version(s) of Spring Integration are you seeing this issue?

Reproduced against main (7.2.0-SNAPSHOT). The affected code is identical in v6.0.0, v6.3.0 and v7.0.0.

Describe the bug

When a GET (or MGET) gateway is configured with a local-directory-expression rather than a fixed local-directory, the local directory is created per message on the calling thread:

private File generateLocalDirectory(Message<?> message, @Nullable String remoteDirectory) {
	...
	File localDir = ExpressionUtils.expressionToFile(this.localDirectoryExpression, evaluationContext, message,
			"Local Directory");
	if (!localDir.exists()) {
		Assert.isTrue(localDir.mkdirs(), () -> "Failed to make local directory: " + localDir);
	}
	return localDir;
}

This check-then-act sequence is not atomic. When two or more threads concurrently deliver messages resolving to the same, not-yet-existing local directory, all of them can pass !localDir.exists(), but only one File.mkdirs() wins. File.mkdirs() returns false when the directory already exists by the time it runs, so the losing threads fail with:

java.lang.IllegalArgumentException: Failed to make local directory: /path/to/local/subdir
	at org.springframework.util.Assert.isTrue(Assert.java:136)
	at org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway.generateLocalDirectory(AbstractRemoteFileOutboundGateway.java:1431)
	at org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway.get(AbstractRemoteFileOutboundGateway.java:1175)

even though the directory exists at that point and the transfer could simply proceed.

The eager creation in afterPropertiesSet() does not cover this case, because it is deliberately restricted to a ValueExpression:

Assert.notNull(this.localDirectoryExpression, "localDirectory must not be null");
if (this.localDirectoryExpression instanceof ValueExpression) {
	setupLocalDirectory();
}

So with a message-dependent expression the directory is never pre-created, and creation relies entirely on the per-message path above. Any multi-threaded flow (for example a poller with a task executor, or an executor-backed channel in front of the gateway) that fetches several remote files into the same dynamically resolved local directory is affected.

This is the same defect class as GH-11253, fixed for FileWritingMessageHandler in GH-11254.

To Reproduce

Add the following test to RemoteFileOutboundGatewayTests (eight threads, 50 rounds, a new local directory per round). Against current main it fails in the first round:

@Test
public void testGetConcurrentCreateSameLocalDirectory(@TempDir File localRoot) throws Exception {
	SessionFactory sessionFactory = mock(SessionFactory.class);
	TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway(sessionFactory, "get", "payload");
	// A non-ValueExpression skips the eager setupLocalDirectory() in afterPropertiesSet(),
	// so the directory is created per message on the calling thread.
	gw.setLocalDirectoryExpression(
			new FunctionExpression<Message<?>>((message) -> message.getHeaders().get("localDir")));
	gw.afterPropertiesSet();
	when(sessionFactory.getSession()).thenReturn(new TestSession() {

		@Override
		public TestLsEntry[] list(String path) {
			return new TestLsEntry[] {
					new TestLsEntry(path, 1234, false, false, 12345, "-rw-r--r--")
			};
		}

		@Override
		public void read(String source, OutputStream outputStream) throws IOException {
			outputStream.write("testfile".getBytes(StandardCharsets.UTF_8));
		}

	});

	int concurrency = 8;
	ExecutorService executorService = Executors.newFixedThreadPool(concurrency);
	AtomicReference<Throwable> failure = new AtomicReference<>();
	try {
		for (int round = 0; round < 50; round++) {
			File localDirectory = new File(localRoot, "local-" + round);
			CountDownLatch getsDone = new CountDownLatch(concurrency);
			for (int i = 0; i < concurrency; i++) {
				// A distinct remote file per thread; only the destination directory is shared.
				Message<String> message = MessageBuilder.withPayload("f" + i)
						.setHeader("localDir", localDirectory.getAbsolutePath())
						.build();
				executorService.execute(() -> {
					try {
						gw.handleRequestMessage(message);
					}
					catch (Throwable ex) { // NOSONAR - the race surfaces as an unchecked exception
						failure.compareAndSet(null, ex);
					}
					finally {
						getsDone.countDown();
					}
				});
			}
			assertThat(getsDone.await(10, TimeUnit.SECONDS)).isTrue();
			assertThat(failure.get()).isNull();
			for (int i = 0; i < concurrency; i++) {
				assertThat(new File(localDirectory, "f" + i)).exists();
			}
		}
	}
	finally {
		executorService.shutdownNow();
	}
}

Expected behavior

Creation of the local directory should be idempotent and safe under concurrency. java.nio.file.Files.createDirectories(Path) provides exactly these semantics - it does not fail when the directory already exists, including when it was created concurrently - and reports the actual cause through IOException when creation genuinely fails:

if (!localDir.exists()) {
	try {
		Files.createDirectories(localDir.toPath());
	}
	catch (IOException ex) {
		throw new IllegalArgumentException("Failed to make local directory: " + localDir, ex);
	}
}

This mirrors the fix applied to FileWritingMessageHandler in GH-11254 and keeps the exception type and message unchanged. With it applied locally, :spring-integration-file:check passes (332 tests, 0 failures) and the test above passes; reverting only the production change makes that single test fail again.

PropertiesPersistingMetadataStore already uses the concurrency-tolerant form (!baseDir.mkdirs() && !baseDir.exists()), so the idiom is established in the codebase.

Sample

The reproducer above is self-contained within the existing RemoteFileOutboundGatewayTests harness. I have the fix and the test ready and will open a pull request against this issue.