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

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;
}

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

@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();
	}
}

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