GH-3507: Fix Tail producer for proper command

Fixes #3507

The `OSDelegatingFileTailingMessageProducer` passing command string to `Runtime.getRuntime().exec()`
may cause problems if spaces (and other special characters) are used in the filename.

* Use an array for command and its options to let the target `Runtime` to parse and
execute it properly

**Cherry-pick to 5.4.x, 5.3.x & 5.2.x**

# Conflicts:
#	spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java

Author: trgpa

spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -33,22 +33,28 @@
  * @author Gary Russell
  * @author Gavin Gray
  * @author Ali Shahbour
+ * @author Artem Bilan
+ * @author Trung Pham
+ *
  * @since 3.0
  *
  */
 public class OSDelegatingFileTailingMessageProducer extends FileTailingMessageProducerSupport
 		implements SchedulingAwareRunnable {
 
-	private volatile Process nativeTailProcess;
 
 	private volatile String options = "-F -n 0";
 
 	private volatile String command = "ADAPTER_NOT_INITIALIZED";
 
-	private volatile boolean enableStatusReader = true;
+	private volatile String[] tailCommand;
+
+	private volatile Process nativeTailProcess;
 
 	private volatile BufferedReader stdOutReader;
 
+	private volatile boolean enableStatusReader = true;
+
 	public void setOptions(String options) {
 		if (options == null) {
 			this.options = "";
@@ -92,8 +98,13 @@ protected void onInit() {
 	protected void doStart() {
 		super.doStart();
 		destroyProcess();
-		this.command = "tail " + this.options + " " + this.getFile().getAbsolutePath();
-		this.getTaskExecutor().execute(this::runExec);
+		String[] tailOptions = this.options.split("\\s+");
+		this.tailCommand = new String[tailOptions.length + 2];
+		this.tailCommand[0] = "tail";
+		this.tailCommand[this.tailCommand.length - 1] = getFile().getAbsolutePath();
+		System.arraycopy(tailOptions, 0, this.tailCommand, 1, tailOptions.length);
+		this.command = String.join(" ", this.tailCommand);
+		getTaskExecutor().execute(this::runExec);
 	}
 
 	@Override
@@ -114,20 +125,18 @@ private void destroyProcess() {
 	 * Exec the native tail process.
 	 */
 	private void runExec() {
-		this.destroyProcess();
-		if (logger.isInfoEnabled()) {
-			logger.info("Starting tail process");
-		}
+		destroyProcess();
+		logger.info("Starting tail process");
 		try {
-			Process process = Runtime.getRuntime().exec(this.command);
+			Process process = Runtime.getRuntime().exec(this.tailCommand);
 			BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()));
 			this.nativeTailProcess = process;
-			this.startProcessMonitor();
+			startProcessMonitor();
 			if (this.enableStatusReader) {
 				startStatusReader();
 			}
 			this.stdOutReader = reader;
-			this.getTaskExecutor().execute(this);
+			getTaskExecutor().execute(this);
 		}
 		catch (IOException e) {
 			throw new MessagingException("Failed to exec tail command: '" + this.command + "'", e);
@@ -139,14 +148,13 @@ private void runExec() {
 	 * Runs a thread that waits for the Process result.
 	 */
 	private void startProcessMonitor() {
-		this.getTaskExecutor().execute(() -> {
-			Process process = OSDelegatingFileTailingMessageProducer.this.nativeTailProcess;
-			if (process == null) {
-				if (logger.isDebugEnabled()) {
-					logger.debug("Process destroyed before starting process monitor");
-				}
-				return;
-			}
+		getTaskExecutor()
+				.execute(() -> {
+					Process process = OSDelegatingFileTailingMessageProducer.this.nativeTailProcess;
+					if (process == null) {
+						logger.debug("Process destroyed before starting process monitor");
+						return;
+					}
 
 			int result = Integer.MIN_VALUE;
 			try {

spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -53,6 +53,7 @@
  * @author Gavin Gray
  * @author Artem Bilan
  * @author Ali Shahbour
+ * @author Trung Pham
  *
  * @since 3.0
  */
@@ -183,7 +184,7 @@ private void testGuts(FileTailingMessageProducerSupport adapter, String field)
 		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
 		taskScheduler.afterPropertiesSet();
 		adapter.setTaskScheduler(taskScheduler);
-		final List<FileTailingEvent> events = new ArrayList<FileTailingEvent>();
+		final List<FileTailingEvent> events = new ArrayList<>();
 		adapter.setApplicationEventPublisher(event -> {
 			FileTailingEvent tailEvent = (FileTailingEvent) event;
 			logger.debug(event);
@@ -232,6 +233,8 @@ private void testGuts(FileTailingMessageProducerSupport adapter, String field)
 		}
 
 		assertThat(events.size()).isGreaterThanOrEqualTo(1);
+
+		taskScheduler.destroy();
 	}
 
 	private void waitForField(FileTailingMessageProducerSupport adapter, String field) throws Exception {
@@ -248,4 +251,38 @@ private void waitForField(FileTailingMessageProducerSupport adapter, String fiel
 		fail("adapter failed to start");
 	}
 
+	@Test
+	@TailAvailable
+	public void canHandleFilenameHavingSpecialCharacters() throws Exception {
+		File file = File.createTempFile("foo bar", " -c 1");
+		file.delete();
+
+		OSDelegatingFileTailingMessageProducer adapter = new OSDelegatingFileTailingMessageProducer();
+		adapter.setOptions(TAIL_OPTIONS_FOLLOW_NAME_ALL_LINES);
+		adapter.setFile(file);
+		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.afterPropertiesSet();
+		adapter.setTaskScheduler(taskScheduler);
+		QueueChannel outputChannel = new QueueChannel();
+		adapter.setOutputChannel(outputChannel);
+		adapter.setTailAttemptsDelay(500);
+		adapter.setBeanFactory(mock(BeanFactory.class));
+		adapter.afterPropertiesSet();
+
+		adapter.start();
+		waitForField(adapter, "stdOutReader");
+
+		FileOutputStream fos = new FileOutputStream(file);
+		fos.write(("hello foobar\n").getBytes());
+		fos.close();
+
+		Message<?> message = outputChannel.receive(10000);
+		assertThat(message).as("expected a non-null message").isNotNull();
+		assertThat(message.getPayload()).isEqualTo("hello foobar");
+
+		adapter.stop();
+		file.delete();
+		taskScheduler.destroy();
+	}
+
 }

spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailRule.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -24,6 +24,7 @@
 import java.io.OutputStream;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
@@ -75,9 +76,6 @@ public void evaluate() {
 	}
 
 	private boolean tailWorksOnThisMachine() {
-		if (tmpDir.contains(":")) {
-			return false;
-		}
 		File testDir = new File(tmpDir, "FileTailingMessageProducerTests");
 		testDir.mkdir();
 		final File file = new File(testDir, "foo");
@@ -88,19 +86,24 @@ private boolean tailWorksOnThisMachine() {
 			fos.close();
 			final AtomicReference<Integer> c = new AtomicReference<>();
 			final CountDownLatch latch = new CountDownLatch(1);
-			Future<Process> future = Executors.newSingleThreadExecutor().submit(() -> {
-				final Process process = Runtime.getRuntime().exec(commandToTest + " " + file.getAbsolutePath());
-				Executors.newSingleThreadExecutor().execute(() -> {
-					try {
-						c.set(process.getInputStream().read());
-						latch.countDown();
-					}
-					catch (IOException e) {
-						logger.error("Error reading test stream", e);
-					}
-				});
-				return process;
-			});
+			ExecutorService newSingleThreadExecutor = Executors.newSingleThreadExecutor();
+			Future<Process> future =
+					newSingleThreadExecutor.submit(() -> {
+						final Process process = Runtime.getRuntime().exec(commandToTest + " " + file.getAbsolutePath());
+						ExecutorService executorService = Executors.newSingleThreadExecutor();
+						executorService.execute(() -> {
+							try {
+								c.set(process.getInputStream().read());
+								latch.countDown();
+							}
+							catch (IOException e) {
+								logger.error("Error reading test stream", e);
+							}
+						});
+						executorService.shutdown();
+						return process;
+					});
+			newSingleThreadExecutor.shutdown();
 			try {
 				Process process = future.get(10, TimeUnit.SECONDS);
 				if (latch.await(10, TimeUnit.SECONDS)) {