[Bug]: [Stream Lookup] Proposal: Use threads + CountDownLatch to eliminate deadlocks

[Rookie625]

Apache Hop version?

2.18.0

Java version?

21

Operating system

Windows

What happened?

Problem Statement

The Stream Lookup transform is known to cause pipeline deadlocks when processing large datasets with split/rejoin streams [0][1]. The root cause is that when one input stream is blocked due to a full rowset buffer (e.g., after a Group By transform), Stream Lookup waits for that stream while the upstream transform cannot proceed, creating a cyclic dependency.

Current official solutions [2] require users to manually adjust pipeline design (increase rowset size, separate streams, split pipelines, or use Blocking transforms). These are workarounds, not root-cause fixes.

Proposed Solution

Use separate threads (regular threads or virtual threads, depending on the runtime environment) to read each info stream independently and concurrently, with a CountDownLatch to coordinate the main thread. This approach eliminates deadlocks because each stream is consumed in its own thread, preventing buffer contention between upstream transforms.

Pseudo‑code (using regular threads for compatibility with Java 11):

private boolean loadInfoStream(IStream stream, CountDownLatch latch, List<Exception> errors) {
    Thread loader = new Thread(() -> {
        try {
            IRowSet rowSet = findInputRowSet(stream.getTransformName());
            Object[] row = getRowFrom(rowSet);
            while (row != null && !isStopped()) {
                // Add to cache (needs thread-safe collection)
                data.cache.add(row);
                row = getRowFrom(rowSet);
            }
        } catch (Exception e) {
            errors.add(e);
        } finally {
            latch.countDown();
        }
    });
    loader.setName("StreamLookup-Loader-" + stream.getTransformName());
    loader.setUncaughtExceptionHandler((t, e) -> errors.add(e));
    loader.start();
    return true;
}

// In  before processingrows():
CountDownLatch latch = new CountDownLatch(infoStreams.size());
List<Exception> errors = Collections.synchronizedList(new ArrayList<>());
for (IStream stream : infoStreams) {
    loadInfoStream(stream, latch, errors);
}
latch.await(); // wait for all loading threads to finish
if (!errors.isEmpty()) {
    throw new HopException("Failed to load lookup data", errors.get(0));
}

Key benefits:

No deadlock – each stream is read independently, no circular blocking.

Transparent to users – no pipeline redesign needed.

Compatible with Java 11 – regular threads work, virtual threads can be used as an optimisation if Java 21+ is available.

Root Cause – A Real Scenario

Consider a simple pipeline:

• Data Input produces 20,000 rows, split into two streams.
• Transform1 and Transform2 each receive 10,000 rows (but could be more).
• Both write into separate rowsets (Rowset1 and Rowset2), each with a default capacity of 10,000 rows.
• StreamLookup consumes these rowsets sequentially with a single thread (first Rowset1, then Rowset2).

What happens:

1. Both transforms write 10,000 rows into their respective rowsets — both are now full.
2. StreamLookup starts reading Rowset1 first.
3. Data Input continues to send more rows to both transforms.
4. Transform1 keeps writing (because StreamLookup is consuming Rowset1, freeing space).
5. Transform2 tries to write the 10,001st row into Rowset2, but the rowset is full and no one is consuming it → Transform2 blocks.
6. StreamLookup is still busy consuming Rowset1 and may never get to Rowset2.
7. Result: Deadlock — Transform2 waits for StreamLookup to consume Rowset2, StreamLookup is stuck on Rowset1, and the entire pipeline is hung.

This is not a theoretical problem — it is a reproducible deadlock caused by the single‑threaded consumer pattern in StreamLookup.

References
[0] https://hop.apache.org/manual/2.18.0/how-to-guides/avoiding-deadlocks.html

[1] #3740

[2] https://hop.incubator.apache.org/manual/2.18.0/pipeline/transforms/streamlookup.html

es.

Issue Priority

Priority: 2

Issue Component

Component: Transforms

[mattcasters]

It's not really a bug but a documented behavior with circular dependencies.
Loading all data into memory is perhaps less than ideal in all scenario. There's a reason why we don't have infinite buffers between transforms and that is obviously to avoid running out of memory.

That being said, if we could detect the circular dependency we could opt to serialize rows to disk to get around the limitation as a generic solution. I guess that shifts the problem to detecting that there is a circular dependency at play.

In any case, my preference would be to avoid starting extra threads and possibly causing out of memory errors and to favor a more generic solution for the problem.

[hansva]

I'm also more in favour of detecting the situation and giving a clear error message (avoid the infinite lock). We have documented workarounds for this. Or we could even suggest inserting a transform to handle this (block).

[mattcasters]

I would appreciate an extra few pairs of eyes on #7854. I propose to give this some more mileage until 2.20.

[Rookie625]

It's not really a bug but a documented behavior with circular dependencies. Loading all data into memory is perhaps less than ideal in all scenario. There's a reason why we don't have infinite buffers between transforms and that is obviously to avoid running out of memory.

That being said, if we could detect the circular dependency we could opt to serialize rows to disk to get around the limitation as a generic solution. I guess that shifts the problem to detecting that there is a circular dependency at play.

In any case, my preference would be to avoid starting extra threads and possibly causing out of memory errors and to favor a more generic solution for the problem.

Thank you for taking the time to provide such a thorough and candid explanation.

I completely see your point. Looking at this from the engine's perspective, it's absolutely not a bug—it's a conscious trade-off between performance and stability. The decision to avoid infinite buffers is precisely what prevents the engine from collapsing under memory pressure, and that foundation is something I fully respect.

My initial observation came from a different angle—more from the user experience side, where encountering a silent hang can be confusing. But I now realize that what feels like a "limitation" from the outside is actually a deliberate architectural safeguard from the inside. It's a classic case of two valid perspectives looking at the same problem.

Your idea about detecting circular dependencies and spilling to disk as a generic solution is particularly insightful. It strikes me as an elegant long-term direction that doesn't compromise the core principle of memory safety. I also appreciate your preference for avoiding extra threads that might invite OOM errors—it reinforces that stability should always come first.

This exchange has given me a much deeper appreciation for the design choices behind the scenes. Sometimes the most robust systems are the ones that refuse to compromise on their fundamental constraints, even if it means occasionally surprising the user.

Thanks again for sharing your thoughts—I've learned a lot from this discussion.