GH-11277: Fix local directory creation race in remote file GET

Fixes: #11277

`generateLocalDirectory()` used a non-atomic `exists()`/`mkdirs()` sequence.
With a `local-directory-expression` the eager `setupLocalDirectory()` in
`afterPropertiesSet()` is skipped, since it only runs for a `ValueExpression`,
so the directory is created per message on the calling thread. Concurrent
messages resolving to the same, not-yet-existing local directory raced on
`File.mkdirs()`: the losing threads received `false` and failed with
`IllegalArgumentException: Failed to make local directory: [...]` although the
directory existed at that point.

* Treat `false` from `mkdirs()` as a failure only when the directory is still
  absent - the concurrency-tolerant idiom already used by
  `PropertiesPersistingMetadataStore`

`Files.createDirectories()`, as used by `FileWritingMessageHandler` since
GH-11254, is race-tolerant as well, but it makes an `IOException` the root cause
of the reported `IllegalArgumentException`. `FtpServerOutboundTests` and
`SftpServerOutboundTests` assert that root cause, so the thrown exception is
kept unchanged in type, message and cause instead.

Signed-off-by: Jiwoo Lee <dlwldn30@naver.com>

Author: dlwldn30

spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java

@@ -77,6 +77,7 @@
  * @author Artem Bilan
  * @author Mauro Molinari
  * @author Jooyoung Pyoung
+ * @author Jiwoo Lee
  *
  * @since 2.1
  */
@@ -1428,7 +1429,9 @@ private File generateLocalDirectory(Message<?> message, @Nullable String remoteD
 		File localDir = ExpressionUtils.expressionToFile(this.localDirectoryExpression, evaluationContext, message,
 				"Local Directory");
 		if (!localDir.exists()) {
-			Assert.isTrue(localDir.mkdirs(), () -> "Failed to make local directory: " + localDir);
+			// mkdirs() returns false when a concurrent caller has created the directory in the meantime
+			Assert.isTrue(localDir.mkdirs() || localDir.exists(),
+					() -> "Failed to make local directory: " + localDir);
 		}
 		return localDir;
 	}

spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java

@@ -23,13 +23,18 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.util.ArrayList;
 import java.util.Calendar;
 import java.util.Collection;
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.jupiter.api.Test;
@@ -38,6 +43,7 @@
 
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.integration.expression.FunctionExpression;
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.filters.AbstractSimplePatternFileListFilter;
 import org.springframework.integration.file.remote.AbstractFileInfo;
@@ -72,6 +78,7 @@
  * @author Liu Jiong
  * @author Artem Bilan
  * @author Jooyoung Pyoung
+ * @author Jiwoo Lee
  *
  * @since 2.1
  */
@@ -809,6 +816,67 @@ public void read(String source, OutputStream outputStream) throws IOException {
 		out.delete();
 	}
 
+	@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();
+		}
+	}
+
 	@Test
 	public void testRm() throws Exception {
 		SessionFactory sessionFactory = mock(SessionFactory.class);