org.easytesting
fest-reflect
diff --git a/flume-ng-core/src/main/java/org/apache/flume/source/ExecSource.java b/flume-ng-core/src/main/java/org/apache/flume/source/ExecSource.java
index c99e8dd5f6..17b03c5814 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/source/ExecSource.java
+++ b/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(
diff --git a/flume-ng-core/src/test/java/org/apache/flume/source/TestExecSource.java b/flume-ng-core/src/test/java/org/apache/flume/source/TestExecSource.java
index d8d369f6bc..e007ff7337 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/source/TestExecSource.java
+++ b/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.
+ *
+ * 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 {