This is an automated email from the ASF dual-hosted git repository.
ppkarwasz pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/logging-flume.git
The following commit(s) were added to refs/heads/trunk by this push:
new 373de83e De-flake `TestExecSource` on Windows (#491)
373de83e is described below
commit 373de83efb1d6a2b4feb335377435fc97464ba03
Author: Piotr P. Karwasz <[email protected]>
AuthorDate: Tue Aug 25 13:46:29 2026 +0200
De-flake `TestExecSource` on Windows (#491)
Fixed sleeps lose to PowerShell start-up times on loaded CI runners:
await the source counter instead.
`Get-Content -Wait` keeps the process alive like `tail -f`,
so the batch timeout, not the EOF flush, delivers the events.
Writing the input file before the source starts removes the race
with the one-shot read of the old command.
Assisted-By: Claude Fable 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_016zsFn1f2B32vcNoCeZSfn7
---
flume-ng-core/pom.xml | 6 +++
.../java/org/apache/flume/source/ExecSource.java | 6 +++
.../org/apache/flume/source/TestExecSource.java | 60 ++++++++++++++--------
3 files changed, 50 insertions(+), 22 deletions(-)
diff --git a/flume-ng-core/pom.xml b/flume-ng-core/pom.xml
index e8b6a0d2..fa9857c8 100644
--- a/flume-ng-core/pom.xml
+++ b/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>
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 c99e8dd5..17b03c58 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 class ExecSource extends AbstractSource implements
EventDrivenSource, Con
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 d8d369f6..e007ff73 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.io.InputStreamReader;
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 @@ import org.junit.Test;
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 class TestExecSource {
// 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 class TestExecSource {
Configurables.configure(source, context);
source.start();
- Thread.sleep(2000);
+ awaitEventCount(10);
Transaction transaction = channel.getTransaction();
transaction.begin();
@@ -286,36 +289,41 @@ public class TestExecSource {
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 class TestExecSource {
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 {
