This is an automated email from the ASF dual-hosted git repository.
mattcasters 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 0ee1c6d27b hardening on lineagehub, fixes #7605 (#7607)
0ee1c6d27b is described below
commit 0ee1c6d27baa8c1f36ebb0fb266dfa0e11980668
Author: Hans Van Akelyen <[email protected]>
AuthorDate: Fri Jul 24 09:37:35 2026 +0200
hardening on lineagehub, fixes #7605 (#7607)
---
.../org/apache/hop/lineage/hub/LineageHub.java | 43 ++++++-
.../lineage/hub/LineageHubFlushInterruptTest.java | 126 +++++++++++++++++++++
.../org/apache/hop/lineage/hub/LineageHubTest.java | 36 ++++--
3 files changed, 189 insertions(+), 16 deletions(-)
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 43c0f09aa3..3239811601 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
@@ -124,7 +124,8 @@ public final class LineageHub {
request.done.await(60, TimeUnit.SECONDS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- log.logError("Interrupted while flushing lineage queue", e);
+ log.logDebug(
+ "Lineage flush interrupted during shutdown; pending events will be
drained on stop");
}
}
@@ -150,13 +151,47 @@ public final class LineageHub {
}
}
worker = null;
+ drainRemaining();
shutdownSinks();
+ sinksInitialized.set(false);
+ sinks = List.of();
+ }
+
+ /**
+ * Synchronously dispatch any events still on the queue to the sinks, on the
caller's thread.
+ * Best-effort: only meaningful after the worker has stopped, so no events
are added concurrently.
+ */
+ private void drainRemaining() {
BlockingQueue<Object> q = queue;
- if (q != null) {
+ if (q == null) {
+ return;
+ }
+ if (!sinksInitialized.get() || sinks.isEmpty()) {
q.clear();
+ return;
+ }
+ LineageConfiguration cfg = resolveConfig();
+ List<Object> pending = new ArrayList<>();
+ q.drainTo(pending);
+ List<LineageEvent> batch = new ArrayList<>();
+ for (Object o : pending) {
+ if (o instanceof FlushRequest fr) {
+ if (!batch.isEmpty()) {
+ dispatchBatch(batch);
+ batch.clear();
+ }
+ fr.done.countDown();
+ } else {
+ batch.add((LineageEvent) o);
+ if (batch.size() >= cfg.getBatchMax()) {
+ dispatchBatch(batch);
+ batch.clear();
+ }
+ }
+ }
+ if (!batch.isEmpty()) {
+ dispatchBatch(batch);
}
- sinksInitialized.set(false);
- sinks = List.of();
}
public long getDroppedEventCount() {
diff --git
a/engine/src/test/java/org/apache/hop/lineage/hub/LineageHubFlushInterruptTest.java
b/engine/src/test/java/org/apache/hop/lineage/hub/LineageHubFlushInterruptTest.java
new file mode 100644
index 0000000000..a6cc9618c0
--- /dev/null
+++
b/engine/src/test/java/org/apache/hop/lineage/hub/LineageHubFlushInterruptTest.java
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.lineage.hub;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.hop.core.logging.HopLogStore;
+import org.apache.hop.lineage.context.LineageContext;
+import org.apache.hop.lineage.model.LineageEvent;
+import org.apache.hop.lineage.model.LineageEventKind;
+import org.apache.hop.lineage.spi.ILineageSink;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Regression test for the reported LineageHub flush-interrupt bug. When the
pipeline engine is
+ * shutting down and interrupts the thread running the PipelineCompleted
extension point, {@link
+ * LineageHub#flush()} was blocked in {@code queue.put(request)} and surfaced
the interrupt as a
+ * full ERROR stack trace. The fix handles the interrupt gracefully (restore
the flag, debug-level
+ * log) and, crucially, the still-queued events are not lost: {@link
LineageHub#shutdown()} drains
+ * them.
+ */
+class LineageHubFlushInterruptTest {
+
+ @BeforeAll
+ static void initLogStore() {
+ HopLogStore.init();
+ }
+
+ @Test
+ void flushInterruptedDuringShutdownIsGracefulAndLosesNoEvents() throws
Exception {
+ // Capacity 1 so the queue fills quickly; large linger so the worker parks
in the sink.
+ LineageConfiguration cfg = LineageConfiguration.forTesting(true, 1, 1,
30_000L, Set.of());
+
+ Semaphore holdSink = new Semaphore(0);
+ AtomicBoolean firstBatch = new AtomicBoolean(true);
+ AtomicInteger persisted = new AtomicInteger();
+ ILineageSink blocking =
+ batch -> {
+ // Park on the first event so a second event fills the single queue
slot and the flush
+ // then blocks in queue.put(); release on interrupt so shutdown can
proceed.
+ if (firstBatch.compareAndSet(true, false)) {
+ try {
+ holdSink.acquire();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ persisted.addAndGet(batch.size());
+ };
+
+ LineageHub hub = LineageHub.newIsolatedForTesting(cfg, List.of(blocking));
+
+ // Event 1: worker pops it and parks in the sink. Event 2: fills the
single queue slot.
+ hub.emit(sampleEvent());
+ Thread.sleep(150);
+ hub.emit(sampleEvent());
+
+ // The flusher plays the role of
RemotePipelineEngine.fireExecutionFinishedListeners ->
+ // LineageHubPipelineCompletedXp.flushQuietly -> flush():
queue.put(FlushRequest) blocks.
+ AtomicBoolean interruptFlagRestored = new AtomicBoolean(false);
+ int bufferBefore = HopLogStore.getAppender().getLastBufferLineNr();
+ Thread flusher =
+ new Thread(
+ () -> {
+ hub.flushQuietly();
+
interruptFlagRestored.set(Thread.currentThread().isInterrupted());
+ },
+ "repro-pipeline-finish-thread");
+ flusher.start();
+
+ // Give the flusher time to reach the blocked queue.put(), then interrupt
it the way the engine
+ // shutdown does.
+ Thread.sleep(300);
+ flusher.interrupt();
+ flusher.join(5_000);
+
+ // The interrupted flush must restore the interrupt flag...
+ assertTrue(
+ interruptFlagRestored.get(), "flush() should restore the interrupt
flag after catching");
+
+ // ...and must NOT surface an ERROR stack trace for the interrupt.
+ String loggedSinceFlush =
+ HopLogStore.getAppender().getBuffer(null, true,
bufferBefore).toString();
+ assertFalse(
+ loggedSinceFlush.contains("Interrupted while flushing lineage queue"),
+ "interrupted flush must no longer be logged as an ERROR; was:\n" +
loggedSinceFlush);
+ assertFalse(
+ loggedSinceFlush.contains("InterruptedException"),
+ "interrupted flush must not dump an InterruptedException stack trace;
was:\n"
+ + loggedSinceFlush);
+
+ // Now the process shuts down. The two emitted events were never lost by
the failed flush and
+ // are drained by shutdown() rather than dropped.
+ holdSink.release(20);
+ hub.shutdown();
+ assertEquals(
+ 2, persisted.get(), "both emitted events must be delivered despite the
interrupted flush");
+ }
+
+ private static LineageEvent sampleEvent() {
+ return LineageEvent.of(LineageEventKind.RUN_LIFECYCLE,
LineageContext.empty(), null);
+ }
+}
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 e99b5d2f39..9d16a71de8 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
@@ -27,6 +27,7 @@ import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.logging.HopLogStore;
@@ -334,29 +335,40 @@ class LineageHubTest {
}
@Test
- void shutdownDiscardsPendingEventsAndUnblocksSink() throws Exception {
+ void shutdownDeliversPendingEventsInsteadOfDroppingThem() throws Exception {
+ // Regression guard for lineage loss on shutdown: events still queued when
the worker stops must
+ // be dispatched by shutdown(), not silently cleared. A short linger keeps
the worker parked in
+ // poll() so the events are genuinely still queued when shutdown() is
called.
LineageConfiguration cfg = LineageConfiguration.forTesting(true, 100, 1,
30_000L, Set.of());
- Semaphore hold = new Semaphore(0);
+ Semaphore holdFirst = new Semaphore(0);
+ AtomicBoolean firstBatch = new AtomicBoolean(true);
AtomicInteger acceptCount = new AtomicInteger();
ILineageSink sink =
batch -> {
- acceptCount.addAndGet(batch.size());
- try {
- hold.acquire();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
+ // Park on the first event so the remaining events pile up on the
queue while the worker
+ // is busy; release on interrupt so shutdown's worker-stop is not
blocked.
+ if (firstBatch.compareAndSet(true, false)) {
+ try {
+ holdFirst.acquire();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
}
+ acceptCount.addAndGet(batch.size());
};
LineageHub hub = LineageHub.newIsolatedForTesting(cfg, List.of(sink));
- for (int i = 0; i < 4; i++) {
- hub.emit(sampleEvent());
+ hub.emit(sampleEvent());
+ Thread.sleep(150); // worker pops event 1 and parks in the sink
+ for (int i = 0; i < 3; i++) {
+ hub.emit(sampleEvent()); // these queue up while the worker is parked
}
- Thread.sleep(150);
long start = System.currentTimeMillis();
hub.shutdown();
long elapsed = System.currentTimeMillis() - start;
- assertTrue(elapsed < 5_000, "shutdown took " + elapsed + "ms");
- assertEquals(1, acceptCount.get());
+ assertTrue(elapsed < 10_000, "shutdown took " + elapsed + "ms");
+ // All four events are delivered: whatever the worker had not yet
processed is drained by
+ // shutdown() on the caller's thread rather than being dropped.
+ assertEquals(4, acceptCount.get());
}
@Test
