[C++][Acero] Asofjoin does not propagate pause upstream

[gitmodimo]

Describe the bug, including details regarding any error messages, version, and platform.

Current implementation of asofjoin node ignores pause requests from downstream. This can lead to big memory consumption when downstream nodes process data slower than the join node.

I created PR46140 that throttles join processing and input queues propagate pauses upstream.

Component(s)

C++

[TheAbry]

any updates on this? have noticed large amounts of memory consumption when using asof join with a downstreamer to_reader() with use_threads = True; even when I do not consume the batches, I see memory usage rising (and presumably the data being computed) unless I use to_reader with use_threads=False.

[gitmodimo]

The fix is pretty straightforward. It is still WIP mostly due to the tests that are flaky. I have trouble making it work on mac.
There is another issue that is amplified by this path - asof_join split batches only when input gets starved. This also leads to memory runaway. I am experimenting with keeping left batches boundaries and it seems work nice(will post PR soon).
@westonpace can you take a look?

[gitmodimo]

Turns out that there was concurrent access to future. After fix all tests are successful.