This is an automated email from the ASF dual-hosted git repository.
hansva 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 06c154185a IT fixes (#8451)
06c154185a is described below
commit 06c154185a7c9d559afd683a72aafc5fe4d1e09b
Author: Hans Van Akelyen <[email protected]>
AuthorDate: Fri Sep 18 16:55:29 2026 +0200
IT fixes (#8451)
Daily integration-test failures:
- minio/minio is gone from Docker Hub: pull the pinned release from
quay.io in the minio and hop_server compose files
- OpenLineage sink died with NoClassDefFoundError (YAMLFactory): the
plugin lib now only leaves out the Jackson modules lib/core ships
(core/databind/annotations, avro, xml, jsr310); jackson-dataformat-yaml
is version-managed with the rest of Jackson
- LineageHub: a sink that fails to load no longer leaves the hub
"running" with a dead dispatcher, which stalled every pipeline and
workflow completion for the 60s flush timeout
- samples: skip the databasevaluevalidation-basic template sample
- remove sequencing hops into Data grid / Get file names that the
unconsumed-main-input guard now rejects (IT 0036, azure 0007)
---
.../integration-tests-hop_server.yaml | 2 +-
.../integration-tests/integration-tests-minio.yaml | 2 +-
.../org/apache/hop/lineage/hub/LineageHub.java | 51 ++++++++++++++++++++--
.../org/apache/hop/lineage/hub/LineageHubTest.java | 38 ++++++++++++++++
integration-tests/azure/0007-get-file-names.hpl | 5 ---
.../samples/read-samples-build-hop-run.hpl | 4 ++
.../0036-actionpipeline-prepare-2rows.hpl | 5 ---
lib/pom.xml | 5 +++
plugins/pom.xml | 4 +-
9 files changed, 98 insertions(+), 18 deletions(-)
diff --git a/docker/integration-tests/integration-tests-hop_server.yaml
b/docker/integration-tests/integration-tests-hop_server.yaml
index aa9a65003a..3d046a3696 100644
--- a/docker/integration-tests/integration-tests-hop_server.yaml
+++ b/docker/integration-tests/integration-tests-hop_server.yaml
@@ -96,7 +96,7 @@ services:
# the client project's metadata-client-only folder, so it only ever reaches
the server inside the
# export ZIP - which is what the test is about.
minio:
- image: minio/minio
+ image: quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z
hostname: minio
environment:
MINIO_REGION: "local"
diff --git a/docker/integration-tests/integration-tests-minio.yaml
b/docker/integration-tests/integration-tests-minio.yaml
index ee39228264..0629b1e80a 100644
--- a/docker/integration-tests/integration-tests-minio.yaml
+++ b/docker/integration-tests/integration-tests-minio.yaml
@@ -27,7 +27,7 @@ services:
- minio
minio:
- image: minio/minio
+ image: quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z
ports:
- "9000:9000"
- "9001:9001"
diff --git a/engine/src/main/java/org/apache/hop/lineage/hub/LineageHub.java
b/engine/src/main/java/org/apache/hop/lineage/hub/LineageHub.java
index 2d3eed67d6..879036ac06 100644
--- a/engine/src/main/java/org/apache/hop/lineage/hub/LineageHub.java
+++ b/engine/src/main/java/org/apache/hop/lineage/hub/LineageHub.java
@@ -59,6 +59,13 @@ public final class LineageHub {
private final AtomicLong droppedEvents = new AtomicLong();
private final AtomicBoolean running = new AtomicBoolean(false);
+
+ /**
+ * Set once the sinks failed to initialize. Without it every later emit()
would start a fresh
+ * worker that fails the same way and logs the same error, once per event.
+ */
+ private final AtomicBoolean sinkInitFailed = new AtomicBoolean(false);
+
private volatile Thread worker;
private volatile List<ILineageSink> sinks = List.of();
@@ -88,6 +95,7 @@ public final class LineageHub {
public void environmentReady() {
LineageConfiguration.invalidate();
sinksInitialized.set(false);
+ sinkInitFailed.set(false);
sinks = List.of();
}
@@ -106,7 +114,7 @@ public final class LineageHub {
*/
public void emit(LineageEvent event) {
LineageConfiguration cfg = resolveConfig();
- if (!cfg.isEnabled()) {
+ if (!cfg.isEnabled() || sinkInitFailed.get()) {
return;
}
ensureWorkerStarted(cfg);
@@ -132,6 +140,11 @@ public final class LineageHub {
FlushRequest request = new FlushRequest();
try {
queue.put(request);
+ if (!running.get()) {
+ // The worker stopped while the request was being queued: nobody will
pick it up.
+ queue.remove(request);
+ return;
+ }
request.done.await(60, TimeUnit.SECONDS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
@@ -165,6 +178,7 @@ public final class LineageHub {
drainRemaining();
shutdownSinks();
sinksInitialized.set(false);
+ sinkInitFailed.set(false);
sinks = List.of();
LineageConfiguration.invalidate();
}
@@ -234,12 +248,26 @@ public final class LineageHub {
}
private void runLoop() {
+ try {
+ dispatchUntilStopped();
+ } finally {
+ // Whatever ends this thread, callers must not keep waiting on a worker
that is gone: with
+ // running still true every flush() would sit out its full timeout, once
per pipeline or
+ // workflow completion, and every event would be queued for nobody.
+ running.set(false);
+ releaseWaitingFlushes();
+ }
+ }
+
+ private void dispatchUntilStopped() {
LineageConfiguration cfg = resolveConfig();
try {
initSinksIfNeeded();
- } catch (HopException e) {
+ } catch (Exception | LinkageError e) {
+ // A sink whose classes do not load (NoClassDefFoundError) ends up here
too. It used to
+ // kill this thread silently, leaving the hub "running" but dispatching
nothing.
+ sinkInitFailed.set(true);
log.logError("Unable to initialize lineage sinks; lineage delivery is
disabled", e);
- running.set(false);
return;
}
@@ -278,7 +306,22 @@ public final class LineageHub {
log.logError("Unexpected error in lineage dispatcher", e);
}
}
- running.set(false);
+ }
+
+ /** Wake up every flush() still waiting on this worker. Only meaningful once
it has stopped. */
+ private void releaseWaitingFlushes() {
+ BlockingQueue<Object> q = queue;
+ if (q == null) {
+ return;
+ }
+ q.removeIf(
+ o -> {
+ if (o instanceof FlushRequest fr) {
+ fr.done.countDown();
+ return true;
+ }
+ return false;
+ });
}
private void processFlush(FlushRequest fr, LineageConfiguration cfg) {
diff --git
a/engine/src/test/java/org/apache/hop/lineage/hub/LineageHubTest.java
b/engine/src/test/java/org/apache/hop/lineage/hub/LineageHubTest.java
index 9d16a71de8..b1d9fdd7ca 100644
--- a/engine/src/test/java/org/apache/hop/lineage/hub/LineageHubTest.java
+++ b/engine/src/test/java/org/apache/hop/lineage/hub/LineageHubTest.java
@@ -150,6 +150,44 @@ class LineageHubTest {
assertEquals(1, initCount.get());
}
+ @Test
+ void sinkThatFailsToLoadDoesNotStallFlush() throws Exception {
+ // The OpenLineage sink once died in init() with a NoClassDefFoundError.
That is not a
+ // HopException, so it killed the dispatcher thread while the hub still
said it was running,
+ // and every pipeline and workflow completion then waited out the full
flush timeout.
+ LineageConfiguration cfg = LineageConfiguration.forTesting(true, 100, 10,
50L, Set.of());
+ AtomicInteger initCount = new AtomicInteger();
+ AtomicInteger accepted = new AtomicInteger();
+ ILineageSink broken =
+ new ILineageSink() {
+ @Override
+ public void init(IVariables variables, ILogChannel log) {
+ initCount.incrementAndGet();
+ throw new
NoClassDefFoundError("com/fasterxml/jackson/dataformat/yaml/YAMLFactory");
+ }
+
+ @Override
+ public void accept(List<LineageEvent> events) {
+ accepted.incrementAndGet();
+ }
+ };
+ LineageHub hub = LineageHub.newIsolatedForTesting(cfg, List.of(broken));
+ hub.emit(sampleEvent());
+
+ long start = System.nanoTime();
+ hub.flush();
+ long millis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
+ assertTrue(millis < 5_000, "flush waited " + millis + " ms on a dead
worker");
+
+ // Later events must not restart a worker that fails the same way, once
per event.
+ hub.emit(sampleEvent());
+ hub.emit(sampleEvent());
+ hub.flush();
+ hub.shutdown();
+ assertEquals(1, initCount.get());
+ assertEquals(0, accepted.get());
+ }
+
@Test
void shutdownInvokesSinkShutdownAndIsIdempotent() throws Exception {
LineageConfiguration cfg = LineageConfiguration.forTesting(true, 100, 10,
50L, Set.of());
diff --git a/integration-tests/azure/0007-get-file-names.hpl
b/integration-tests/azure/0007-get-file-names.hpl
index c9d00f5da3..8e7cd35fa7 100644
--- a/integration-tests/azure/0007-get-file-names.hpl
+++ b/integration-tests/azure/0007-get-file-names.hpl
@@ -38,11 +38,6 @@ limitations under the License.
<notepads>
</notepads>
<order>
- <hop>
- <from>Sparkle</from>
- <to>Get file names</to>
- <enabled>Y</enabled>
- </hop>
<hop>
<from>Get file names</from>
<to>Select values</to>
diff --git a/integration-tests/samples/read-samples-build-hop-run.hpl
b/integration-tests/samples/read-samples-build-hop-run.hpl
index e6bdc4e9bf..1e8ec084d0 100644
--- a/integration-tests/samples/read-samples-build-hop-run.hpl
+++ b/integration-tests/samples/read-samples-build-hop-run.hpl
@@ -555,6 +555,10 @@ var hop_run_cmd = './hop-run.sh -j samples -r ' +
run_config + ' -f ' + filename
<item>table-compare.hpl</item>
<item>requires connection</item>
</line>
+ <line>
+ <item>databasevaluevalidation-basic.hpl</item>
+ <item>template sample: requires connection and target table</item>
+ </line>
<line>
<item>rest-client-slack-conversations-loop.hpl</item>
<item>requires SLACK_BOT_TOKEN with channels:read or
conversations:read</item>
diff --git a/integration-tests/transforms/0036-actionpipeline-prepare-2rows.hpl
b/integration-tests/transforms/0036-actionpipeline-prepare-2rows.hpl
index d89d12cfd1..9fe362a0e7 100644
--- a/integration-tests/transforms/0036-actionpipeline-prepare-2rows.hpl
+++ b/integration-tests/transforms/0036-actionpipeline-prepare-2rows.hpl
@@ -48,11 +48,6 @@ limitations under the License.
<to>Start</to>
<enabled>Y</enabled>
</hop>
- <hop>
- <from>Start</from>
- <to>Data grid</to>
- <enabled>Y</enabled>
- </hop>
</order>
<transform>
<name>Copy rows to result</name>
diff --git a/lib/pom.xml b/lib/pom.xml
index 5641177dce..02cac9557f 100644
--- a/lib/pom.xml
+++ b/lib/pom.xml
@@ -350,6 +350,11 @@
<artifactId>jackson-dataformat-xml</artifactId>
<version>${jackson.version}</version>
</dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.dataformat</groupId>
+ <artifactId>jackson-dataformat-yaml</artifactId>
+ <version>${jackson.version}</version>
+ </dependency>
<dependency>
<groupId>com.fasterxml.jackson.datatype</groupId>
<artifactId>jackson-datatype-jdk8</artifactId>
diff --git a/plugins/pom.xml b/plugins/pom.xml
index f741599abf..0b10ac73b3 100644
--- a/plugins/pom.xml
+++ b/plugins/pom.xml
@@ -178,8 +178,8 @@
<configuration>
<includeScope>runtime</includeScope>
<outputDirectory>${project.build.directory}/hop-plugin-lib</outputDirectory>
-
<excludeGroupIds>org.apache.hop,org.slf4j,org.eclipse.platform,com.fasterxml.jackson.core,com.fasterxml.jackson.dataformat,com.fasterxml.jackson.datatype</excludeGroupIds>
-
<excludeArtifactIds>lombok,jakarta.servlet-api,redshift-jdbc42,clickhouse-jdbc,h2,mssql-jdbc,mysql-connector-j,vertica-jdbc,crate-jdbc,ucanaccess,monetdb-jdbc,jt400,snowflake-jdbc,derbyclient,duckdb_jdbc,hsqldb,postgresql,sqlite-jdbc,${hop.pluginlib.exclude1},${hop.pluginlib.exclude2},${hop.pluginlib.exclude3},${hop.pluginlib.exclude4},${hop.pluginlib.exclude5}</excludeArtifactIds>
+
<excludeGroupIds>org.apache.hop,org.slf4j,org.eclipse.platform,com.fasterxml.jackson.core</excludeGroupIds>
+
<excludeArtifactIds>lombok,jakarta.servlet-api,jackson-dataformat-avro,jackson-dataformat-xml,jackson-datatype-jsr310,redshift-jdbc42,clickhouse-jdbc,h2,mssql-jdbc,mysql-connector-j,vertica-jdbc,crate-jdbc,ucanaccess,monetdb-jdbc,jt400,snowflake-jdbc,derbyclient,duckdb_jdbc,hsqldb,postgresql,sqlite-jdbc,${hop.pluginlib.exclude1},${hop.pluginlib.exclude2},${hop.pluginlib.exclude3},${hop.pluginlib.exclude4},${hop.pluginlib.exclude5}</excludeArtifactIds>
</configuration>
</execution>
</executions>
