This is an automated email from the ASF dual-hosted git repository.
bamaer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new a896af9c76 fix flaky tests, fixes #9155 (#9156)
a896af9c76 is described below
commit a896af9c76e4b6cbb0a7c728c5e569748b13740e
Author: Hans Van Akelyen <[email protected]>
AuthorDate: Sat Oct 31 08:04:24 2026 +0100
fix flaky tests, fixes #9155 (#9156)
* fix flaky tests, fixes #9155
* apply the fix
---
.../hop/junit/rules/RestoreHopEnvironmentExtension.java | 16 ++++++++++++++++
.../vfs/s3/s3common/S3CommonPipedOutputStreamTest.java | 7 ++++++-
.../transforms/watchfiles/WatchFilesIntegrationTest.java | 15 +++++++++++++++
3 files changed, 37 insertions(+), 1 deletion(-)
diff --git
a/core/src/test/java/org/apache/hop/junit/rules/RestoreHopEnvironmentExtension.java
b/core/src/test/java/org/apache/hop/junit/rules/RestoreHopEnvironmentExtension.java
index 271b8ff11a..f3d2516e41 100644
---
a/core/src/test/java/org/apache/hop/junit/rules/RestoreHopEnvironmentExtension.java
+++
b/core/src/test/java/org/apache/hop/junit/rules/RestoreHopEnvironmentExtension.java
@@ -88,6 +88,7 @@ public class RestoreHopEnvironmentExtension implements
BeforeAllCallback, AfterA
}
protected void cleanUp() {
+ resetHopEnvironment();
HopClientEnvironment.reset();
PluginRegistry.getInstance().reset();
MetricsRegistry.getInstance().reset();
@@ -101,6 +102,21 @@ public class RestoreHopEnvironmentExtension implements
BeforeAllCallback, AfterA
ValueMetaFactory.pluginRegistry = PluginRegistry.getInstance();
}
+ /**
+ * The plugin registry and the log store are reset here, so HopEnvironment
must be too: otherwise
+ * a later HopEnvironment.init() in the same JVM believes it is done and
leaves them empty.
+ * HopEnvironment is in hop-engine, which hop-core does not depend on.
+ */
+ private static void resetHopEnvironment() {
+ try {
+
Class.forName("org.apache.hop.core.HopEnvironment").getMethod("reset").invoke(null);
+ } catch (ClassNotFoundException e) {
+ // Not on the class path: nothing to reset
+ } catch (ReflectiveOperationException e) {
+ throw new IllegalStateException("Unable to reset HopEnvironment", e);
+ }
+ }
+
@Override
public void beforeAll(ExtensionContext context) throws Exception {
originalProperties = System.getProperties();
diff --git
a/plugins/tech/aws/src/test/java/org/apache/hop/vfs/s3/s3common/S3CommonPipedOutputStreamTest.java
b/plugins/tech/aws/src/test/java/org/apache/hop/vfs/s3/s3common/S3CommonPipedOutputStreamTest.java
index f6e2ceadf7..6b7ecfee50 100644
---
a/plugins/tech/aws/src/test/java/org/apache/hop/vfs/s3/s3common/S3CommonPipedOutputStreamTest.java
+++
b/plugins/tech/aws/src/test/java/org/apache/hop/vfs/s3/s3common/S3CommonPipedOutputStreamTest.java
@@ -104,7 +104,12 @@ class S3CommonPipedOutputStreamTest {
S3CommonPipedOutputStream out =
new S3CommonPipedOutputStream(fileSystem, "test-bucket", "test-key");
- out.write(new byte[] {1, 2, 3});
+
+ try {
+ out.write(new byte[] {1, 2, 3});
+ } catch (IOException e) {
+ // The upload may already have failed and closed the pipe: close() must
still report it
+ }
assertThrows(
IOException.class,
diff --git
a/plugins/transforms/watchfiles/src/test/java/org/apache/hop/pipeline/transforms/watchfiles/WatchFilesIntegrationTest.java
b/plugins/transforms/watchfiles/src/test/java/org/apache/hop/pipeline/transforms/watchfiles/WatchFilesIntegrationTest.java
index 9f8b1a070b..90b7346e04 100644
---
a/plugins/transforms/watchfiles/src/test/java/org/apache/hop/pipeline/transforms/watchfiles/WatchFilesIntegrationTest.java
+++
b/plugins/transforms/watchfiles/src/test/java/org/apache/hop/pipeline/transforms/watchfiles/WatchFilesIntegrationTest.java
@@ -99,6 +99,9 @@ class WatchFilesIntegrationTest {
for (String name : names) Files.delete(nested.resolve(name));
awaitNames(rows, names, "DELETED");
assertTrue(pipeline.isRunning());
+ // A row reaches the listener before the transform acknowledges it. An
event that is not
+ // acknowledged when the pipeline stops is emitted again after the
restart.
+ awaitAcknowledged("names-" + strategy, 3L * names.size());
} finally {
stop(pipeline);
}
@@ -115,6 +118,18 @@ class WatchFilesIntegrationTest {
}
}
+ private void awaitAcknowledged(String watchId, long expected) throws
Exception {
+ String key =
temporary.resolve("state").toRealPath().resolve(watchId).toString();
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+ WatchFilesDiagnostics.Snapshot snapshot =
WatchFilesDiagnostics.active(key);
+ while ((snapshot == null || snapshot.emitted() < expected) &&
System.nanoTime() < deadline) {
+ Thread.sleep(10);
+ snapshot = WatchFilesDiagnostics.active(key);
+ }
+ assertTrue(
+ snapshot != null && snapshot.emitted() >= expected, "Events not
acknowledged: " + snapshot);
+ }
+
private void awaitNames(
LinkedBlockingQueue<Object[]> rows, java.util.List<String> expected,
String event)
throws Exception {
