zf18634362654 opened a new issue, #7742:
URL: https://github.com/apache/hop/issues/7742
### Apache Hop version?
2.18.1
### Java version?
21
### Operating system
Windows
### What happened?
In the BlockingBatchingRowSet.setDone() method, the call to super.setDone()
is placed before the final batch of data is enqueued. Under multi‑threaded
execution, this can cause a race condition: a downstream consumer may see the
done flag as true before the last data items are actually pushed into the
getArray queue, leading to premature termination and data loss.
We discovered this issue during our stress testing – when the upstream
transform finishes and the downstream hasn't consumed all rows yet, the
downstream block might incorrectly think the stream is already complete and
stop reading.
Steps to Reproduce
Set up a pipeline that uses BlockingBatchingRowSet with a high level of
concurrency (multiple upstream producers and downstream consumers).
Let the upstream producer finish exactly while the downstream consumer is
still processing the last few batches.
In some runs, the downstream consumer will stop early and the final rows are
never processed.
Expected Behavior
All rows should be safely delivered to downstream consumers before the
BlockingBatchingRowSet is marked as done. The done flag should only be set
after the last batch has been enqueued.
Actual Behavior
The done flag is set immediately when super.setDone() is called, even if the
final buffer hasn't been offered to the queue. This can cause downstream
consumers to exit early.
Proposed Fix
Change the order of operations inside setDone(): first enqueue the final
batch (if any), then clear the putArray, and only after that call
super.setDone().
Current code:
text
public void setDone() {
super.setDone();
if (putIndex > 0 && putIndex < size && inputBuffer != null) {
inputBuffer[putIndex] = null; // signal the end of buffer
for (int i = putIndex + 1; i < size; i++) {
inputBuffer[i] = null;
}
getArray.offer(inputBuffer);
}
putArray.clear();
}
Suggested fix (ensure data is enqueued before marking done):
text
public void setDone() {
// 1. Enqueue the last batch first
if (putIndex > 0 && putIndex < size && inputBuffer != null) {
inputBuffer[putIndex] = null; // signal the end of buffer
for (int i = putIndex + 1; i < size; i++) {
inputBuffer[i] = null;
}
getArray.offer(inputBuffer);
}
putArray.clear();
// 2. Mark as done only after the last batch is available to consumers
super.setDone();
}
Additional Note
We would like to thank the Apache Hop community for this great tool. We have
many ETL ideas that we’ve derived from our experience with other industrial ETL
tools (e.g., those used in applied materials), and we would be happy to share
our thoughts and collaborate further. Feel free to reach out if you’d like to
discuss more.
Looking forward to your feedback!
### Issue Priority
Priority: 0
### Issue Component
Component: Transforms
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
