PabloPauling opened a new pull request, #40721:
URL: https://github.com/apache/beam/pull/40721
Fixes #40720.
`ActiveBundle.close()` waited for the `ProcessBundleResponse` and only then
consumed the bundle's output. The inbound queue holds 100 messages, and since
#38407 the Python SDK's send queue is bounded too (100 MB). The SDK does not
complete a bundle until its output has been sent. So a bundle whose output
exceeds both buffers deadlocks: the SDK blocks sending, and the runner blocks
waiting for a response that never comes. On Flink batch, one bundle is a
whole
partition, so this triggers at about 1 GB of output per subtask.
Changes in `close()`:
* It now calls `awaitCompletion()` on the inbound observer before
`MoreFutures.get(response)`, so output is consumed while the SDK is still
producing it.
* Elements embedded in the response are delivered from an asynchronous
`whenCompleteAsync` callback. It has to be async: if the response is
already
complete, a synchronous callback would run on the draining thread and could
block putting into the queue that thread must drain.
* A failed response closes the inbound observer, so the drain does not wait
for
an `is_last` that will never arrive. The response's error stays the primary
exception, and the cancelled-queue exception is added as suppressed. The
existing `handleCleanup*` tests still pass unchanged.
Tests:
* `testOutputLargerThanTheInboundQueueDoesNotDeadlock`: the SDK sends 150
data
messages and only then the response. Times out without the fix.
* `testOutputEmbeddedInAnAlreadyCompletedResponseIsDelivered`: covers the
embedded-elements path.
Also adds a entry under 2.78.0 Bugfixes.
cc @scwhittle (author of #38407)
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
