fix: concurrency bug when doing high volume pull query over java client (alternative)

[lucasbru]

Description

Fix for a race inside vert.x’s RecordParser. Two threads are hitting RecordParserImpl.handleParsing, one is entering from upstream via RecordParser.handle, and one thread is coming from the downstream via ReadStream.resume. Once our PullQueryWriteStream is half empty, we call the drain handler, which calls RecordParser.resume, which synchronously starts parsing new records from the buffer. RecordParser.handleParsing is not thread-safe.

The cause is that the monitor inside PullQueryWriteStream is insufficient to protect from races inside the RecordParser. This makes two changes:

• We run the drain handler as part of the ReadStream context, which ensures that it is race-free with other calls into RecordParser since it's executed as part of the same netty event loop.
• To avoid clogging the event loop with a lot of drain handler calls, we make sure to only call each drain handler exactly once. The PipeImpl will re-register the drain handler when the WriteStream reaches capacity.

Testing done

No behavior changes in existing integration tests. Tested locally.

Reviewer checklist

• Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
• Ensure relevant issues are linked (description should include text like "Fixes #")
• Do these changes have compatibility implications for rollback? If so, ensure that the ksql command version is bumped.

[cla-assistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

[cla-assistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

[lucasbru]

I don't see how this is necessary, since every time we call the drain handlers, we anyways acquire the monitor.

[cadonna]

LGTM!

[AlanConfluent]

It seems like other implementors only allow one handler at a time, and they just clobber any previous one rather than keeping a list, but I couldn't find good documentation on this.

I also think proper callers also reregister after every callback they get, but I couldn't find a very good documentation of this either.

Ultimately, the only user of these APIs are the http response objects, so if we test it out with them, we're good.