De-flake TestExecSource on Windows

flume-ng-core/pom.xml

@@ -98,6 +98,12 @@
       <scope>test</scope>
     </dependency>
 
+    <dependency>
+      <groupId>org.awaitility</groupId>
+      <artifactId>awaitility</artifactId>
+      <scope>test</scope>
+    </dependency>
+
     <dependency>
       <groupId>org.easytesting</groupId>
       <artifactId>fest-reflect</artifactId>

flume-ng-core/src/main/java/org/apache/flume/source/ExecSource.java

@@ -16,6 +16,7 @@
  */
 package org.apache.flume.source;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import java.io.BufferedReader;
@@ -260,6 +261,11 @@ public long getBatchSize() {
         return bufferCount;
     }
 
+    @VisibleForTesting
+    SourceCounter getSourceCounter() {
+        return sourceCounter;
+    }
+
     private static class ExecRunnable implements Runnable {
 
         public ExecRunnable(

flume-ng-core/src/test/java/org/apache/flume/source/TestExecSource.java

@@ -16,6 +16,7 @@
  */
 package org.apache.flume.source;
 
+import static org.awaitility.Awaitility.await;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
@@ -31,6 +32,7 @@
 import java.lang.management.ManagementFactory;
 import java.nio.charset.Charset;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import java.util.regex.Pattern;
 import javax.management.Attribute;
 import javax.management.AttributeList;
@@ -58,9 +60,9 @@
 public class TestExecSource {
 
     private AbstractSource source;
-    private Channel channel = new MemoryChannel();
-    private Context context = new Context();
-    private ChannelSelector rcs = new ReplicatingChannelSelector();
+    private final Channel channel = new MemoryChannel();
+    private final Context context = new Context();
+    private final ChannelSelector rcs = new ReplicatingChannelSelector();
 
     @Before
     public void setUp() {
@@ -101,7 +103,8 @@ public void testProcess() throws InterruptedException, LifecycleException, Event
         // Generates input file with a random data set (10 lines, 200 characters each)
         FileOutputStream outputStream1 = new FileOutputStream(inputFile);
         for (int i = 0; i < 10; i++) {
-            outputStream1.write(RandomStringUtils.randomAlphanumeric(200).getBytes());
+            outputStream1.write(
+                    RandomStringUtils.insecure().nextAlphanumeric(200).getBytes());
             outputStream1.write('\n');
         }
         outputStream1.close();
@@ -116,7 +119,7 @@ public void testProcess() throws InterruptedException, LifecycleException, Event
         Configurables.configure(source, context);
 
         source.start();
-        Thread.sleep(2000);
+        awaitEventCount(10);
         Transaction transaction = channel.getTransaction();
 
         transaction.begin();
@@ -286,36 +289,41 @@ public void testMonitoredCounterGroup()
     public void testBatchTimeout()
             throws InterruptedException, LifecycleException, EventDeliveryException, IOException {
 
-        String filePath = "/tmp/flume-execsource." + Thread.currentThread().getId();
+        File file = File.createTempFile("flume-execsource", null);
+        FileUtils.forceDeleteOnExit(file);
         String eventBody = "TestMessage";
-        FileOutputStream outputStream = new FileOutputStream(filePath);
+
+        // Write the file up front, so the command output does not depend on
+        // when the process starts reading; both commands below print the last
+        // lines of an existing file and then follow it.
+        FileOutputStream outputStream = new FileOutputStream(file);
+        for (int lineNumber = 0; lineNumber < 3; lineNumber++) {
+            outputStream.write((eventBody).getBytes());
+            outputStream.write(String.valueOf(lineNumber).getBytes());
+            outputStream.write('\n');
+        }
+        outputStream.close();
 
         context.put(ExecSourceConfigurationConstants.CONFIG_BATCH_SIZE, "50000");
         context.put(ExecSourceConfigurationConstants.CONFIG_BATCH_TIME_OUT, "750");
         context.put(
                 "shell",
                 SystemUtils.IS_OS_WINDOWS ? "powershell -ExecutionPolicy Unrestricted -command" : "/bin/bash -c");
+        // The process must outlive the batch timeout, so that only the timed
+        // flush can deliver the events.
         context.put(
                 "command",
                 SystemUtils.IS_OS_WINDOWS
-                        ? "Get-Content " + filePath + " | Select-Object -Last 10"
-                        : ("tail -f " + filePath));
+                        ? "Get-Content -Tail 10 -Wait '" + file.getAbsolutePath() + "'"
+                        : ("tail -f " + file.getAbsolutePath()));
 
         Configurables.configure(source, context);
         source.start();
+        awaitEventCount(3);
 
         Transaction transaction = channel.getTransaction();
         transaction.begin();
 
-        for (int lineNumber = 0; lineNumber < 3; lineNumber++) {
-            outputStream.write((eventBody).getBytes());
-            outputStream.write(String.valueOf(lineNumber).getBytes());
-            outputStream.write('\n');
-            outputStream.flush();
-        }
-        outputStream.close();
-        Thread.sleep(1500);
-
         for (int i = 0; i < 3; i++) {
             Event event = channel.take();
             assertNotNull(event);
@@ -326,19 +334,27 @@ public void testBatchTimeout()
         transaction.commit();
         transaction.close();
         source.stop();
-        File file = new File(filePath);
         FileUtils.forceDelete(file);
     }
 
+    /**
+     * Waits until the source accepted the given number of events.
+     *
+     * <p>A fixed sleep is not enough on slow environments, where starting
+     * the child process alone can take several seconds.
+     */
+    private void awaitEventCount(int expected) {
+        await().atMost(30, TimeUnit.SECONDS)
+                .until(() -> ((ExecSource) source).getSourceCounter().getEventAcceptedCount() >= expected);
+    }
+
     private void runTestShellCmdHelper(String shell, String command, String[] expectedOutput)
             throws InterruptedException, LifecycleException, EventDeliveryException, IOException {
         context.put("shell", shell);
         context.put("command", command);
         Configurables.configure(source, context);
         source.start();
-        // Some commands might take longer to complete, specially on Windows
-        // or on slow environments (e.g. Travis CI).
-        Thread.sleep(2500);
+        awaitEventCount(expectedOutput.length);
         Transaction transaction = channel.getTransaction();
         transaction.begin();
         try {