FileWritingMessageHandler: concurrent auto-creation of the same destination directory fails with "Destination directory [...] could not be created." #11253

private static void validateDestinationDirectory(File destinationDirectory, boolean autoCreateDirectory) {
	if (!destinationDirectory.exists() && autoCreateDirectory) {
		Assert.isTrue(destinationDirectory.mkdirs(),
				() -> "Destination directory [" + destinationDirectory + "] could not be created.");
	}
	...

if (!destinationDirectory.exists() && autoCreateDirectory) {
	try {
		Files.createDirectories(destinationDirectory.toPath());
	}
	catch (IOException ex) {
		throw new IllegalArgumentException(
				"Destination directory [" + destinationDirectory + "] could not be created.", ex);
	}
}

import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.file.FileHeaders;
import org.springframework.integration.file.outbound.FileWritingMessageHandler;
import org.springframework.integration.support.MessageBuilder;

public class Repro {

	@EnableIntegration
	static class Config {
	}

	public static void main(String[] args) throws Exception {
		Path base = Files.createTempDirectory("si-race-");

		AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(Config.class);
		FileWritingMessageHandler handler = new FileWritingMessageHandler(
				new SpelExpressionParser().parseExpression("headers['dir']"));
		handler.setBeanFactory(ctx.getBeanFactory());
		handler.setExpectReply(false);
		handler.afterPropertiesSet();

		int threads = 8;
		ExecutorService executor = Executors.newFixedThreadPool(threads);
		try {
			for (int round = 0; round < 500; round++) {
				String dir = base.resolve("sub-" + round).toString();
				CyclicBarrier barrier = new CyclicBarrier(threads);
				List<Future<?>> futures = new ArrayList<>();
				for (int t = 0; t < threads; t++) {
					String fileName = "file-" + t + ".txt";
					futures.add(executor.submit(() -> {
						barrier.await();
						handler.handleMessage(MessageBuilder.withPayload("payload")
								.setHeader("dir", dir)
								.setHeader(FileHeaders.FILENAME, fileName)
								.build());
						return null;
					}));
				}
				for (Future<?> f : futures) {
					try {
						f.get();
					}
					catch (Exception e) {
						Throwable rootCause = e;
						while (rootCause.getCause() != null) {
							rootCause = rootCause.getCause();
						}
						System.out.println("FAILED in round " + round + ": " + rootCause);
						return;
					}
				}
			}
			System.out.println("no failure in 500 rounds");
		}
		finally {
			executor.shutdownNow();
		}
	}

}