fix(arrow/flight/flightsql): drain channel in flightSqlServer.DoGet#437
fix(arrow/flight/flightsql): drain channel in flightSqlServer.DoGet#437zeroshade merged 1 commit intoapache:mainfrom
Conversation
|
Curious, somehow this change is causing a deadlock in the tests. I'll see if I can dig a bit and figure out the cause. |
…ecordWriter in flightSqlServer.DoGet
e7ac840 to
4d9cf7f
Compare
I see what the issue was, and that's my bad. For some reason I did not push the code I intended in the first place. var cc <-chan flight.StreamChunk
for chunk := range cc {
...
}I have amended the commit and pushed the intended version. Sorry for the inconvenience. [EDIT]: [2nd EDIT]: |
|
Nice, i'll take a look at seeing the best way to craft a unit test for this. Thanks for the fix! |
zeroshade
left a comment
There was a problem hiding this comment.
Since I plan on cutting a release, I think I'll get this in now and work out a test for it as a separate PR just to ensure this fix goes out with the release.
|
Thanks for taking care of this 🫡 |
…afe (#615) ### Rationale for this change `StreamChunksFromReader` previously did not observe context cancellation. As a result, if a client disconnected early, the reader could continue producing data indefinitely, potentially blocking on channel sends, leaking `RecordBatch` objects, leaking the reader, and consuming unbounded memory and CPU (this observation triggered this PR). This fix ensures that data streaming promptly stops when the client disconnects. ### What changes are included in this PR? - `StreamChunksFromReader` now accepts a `context.Context`. - Tiny change was made to `DoGet`, to ensure it continues to work with the context-aware `StreamChunksFromReader`. ### Are these changes tested? - To be removed from description: the tests are bit tricky to write, similar to that of #437. Maybe @zeroshade has suggestions? ### Are there any user-facing changes? - `StreamChunksFromReader` now accepts a `context.Context`. --------- Co-authored-by: awakim <arnold.wakim@gmail.com>
Rationale for this change
This PR fixes #435.
If writing to the RecordWriter fails, the current implementation does not drain the channel when an error occurs while writing to the record writer.
What changes are included in this PR?
We add a deferred function that will drain the channel and call
.Release()on the underlyingStreamChunksin order to be able to release the RecordReader.Are these changes tested?
@zeroshade as you suggested, I'll gladly accept a recommendation for the test setup.
Are there any user-facing changes?
No.