You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "PabloPauling (via GitHub)" <gi...@apache.org> on 2026/10/22 08:37:23 UTC
[PR] [Java fn-execution] Drain bundle output before waiting for the ProcessBundleResponse [beam]
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Java fn-execution] Drain bundle output before waiting for the ProcessBundleResponse [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #40721:
URL: https://github.com/apache/beam/pull/40721#issuecomment-6445087306
Thanks!
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Java fn-execution] Drain bundle output before waiting for the ProcessBundleResponse [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle merged PR #40721:
URL: https://github.com/apache/beam/pull/40721
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Java fn-execution] Drain bundle output before waiting for the ProcessBundleResponse [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #40721:
URL: https://github.com/apache/beam/pull/40721#discussion_r4376149724
##########
runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/SdkHarnessClient.java:
##########
@@ -516,22 +516,58 @@ public void close() throws Exception {
} finally {
beamFnDataOutboundAggregator.discard();
}
+ // Consume the bundle's output BEFORE waiting for the ProcessBundleResponse. The inbound
+ // queue is bounded, and the SDK does not finish the bundle until its output has been sent.
+ // Waiting for the response first therefore deadlocks once a single bundle produces more
+ // output than the inbound queue and the SDK's own send buffer hold together: the SDK
+ // blocks sending, the response never arrives, and nothing drains the queue. Batch runners
+ // that hand a whole partition to the SDK as one bundle hit this with ~1 GB of output.
+ boolean inboundDrained = false;
+ Exception drainException = null;
+ if (exception == null && beamFnDataInboundObserver.isPresent()) {
+ BeamFnDataInboundObserver inboundObserver = beamFnDataInboundObserver.get();
+ // Async: when the response is already complete, a synchronous callback would run on
+ // this thread and could block putting elements into the queue this thread must drain.
+ response.whenCompleteAsync(
+ (completedResponse, failure) -> {
+ try {
+ if (failure != null) {
+ // The SDK will not send the rest of the output; unblock the drain below.
+ inboundObserver.close();
+ } else if (completedResponse.hasElements()) {
+ inboundObserver.accept(completedResponse.getElements());
+ }
+ } catch (Exception e) {
+ // Only possible once the drain has failed and cancelled the queue, and that
+ // failure is reported by close().
+ LOG.debug("Dropping the response elements of failed bundle {}", bundleId, e);
+ }
+ });
+ inboundDrained = true;
Review Comment:
can we remove inboundDrained variable? Seems like we can close the inboundObserver unconditionally as it is idempotent.
It might also be clearer to move to closing it on line 554 so it is closer to where we are either awaiting for input or not.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Java fn-execution] Drain bundle output before waiting for the ProcessBundleResponse [beam]
Posted by "PabloPauling (via GitHub)" <gi...@apache.org>.
PabloPauling commented on PR #40721:
URL: https://github.com/apache/beam/pull/40721#issuecomment-6336648447
/gemini review
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Java fn-execution] Drain bundle output before waiting for the ProcessBundleResponse [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #40721:
URL: https://github.com/apache/beam/pull/40721#issuecomment-6336651662
Assigning reviewers:
R: @shunping added as fallback since no labels match configuration
Backup reviewer: Abacn (if the main reviewer is unresponsive, you can escalate to them by tagging their username)
This pull request likely touches a core component ("core" label). Please review with scrutiny.
Note: If you would like to opt out of this review, comment `assign to next reviewer`.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Java fn-execution] Drain bundle output before waiting for the ProcessBundleResponse [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #40721:
URL: https://github.com/apache/beam/pull/40721#discussion_r4346258742
##########
runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/SdkHarnessClient.java:
##########
@@ -546,6 +578,15 @@ public void close() throws Exception {
exception.addSuppressed(e);
}
}
+ // After the response, so that a failed bundle reports the SDK's error rather than the
+ // cancelled queue it caused.
+ if (drainException != null) {
Review Comment:
**🟢 Low**: If `inboundObserver.awaitCompletion()` fails (e.g., a downstream output receiver throws an exception or the data stream closes prematurely) while the SDK's `ProcessBundleResponse` completes normally, `drainException` is non-null but `exception` remains `null` during the `MoreFutures.get(response)` block above (lines 556–567). As a result, `progressHandler.onCompleted`, `checkpointHandler.onCheckpoint`, and `finalizationHandler.requestsFinalization` will still be invoked before `stateRegistration.abort()` is called and `drainException` is thrown.
Consider guarding the completion/checkpoint/finalization callbacks at lines 561–567 with `if (drainException == null)` so that a bundle whose output drain failed does not report completion or register finalization/residuals.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Java fn-execution] Drain bundle output before waiting for the ProcessBundleResponse [beam]
Posted by "PabloPauling (via GitHub)" <gi...@apache.org>.
PabloPauling commented on code in PR #40721:
URL: https://github.com/apache/beam/pull/40721#discussion_r4346294891
##########
runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/SdkHarnessClient.java:
##########
@@ -546,6 +578,15 @@ public void close() throws Exception {
exception.addSuppressed(e);
}
}
+ // After the response, so that a failed bundle reports the SDK's error rather than the
+ // cancelled queue it caused.
+ if (drainException != null) {
Review Comment:
Good point, fixed in 5b2c223. The completion, checkpoint and finalization callbacks now only run when the output drain succeeded. A bundle whose drain failed reports neither residuals nor finalization (its state registration was already aborted). Covered by testFailedOutputDrainDoesNotReportCompletion. The original code also ran these callbacks before draining, so this tightens pre-existing behaviour too.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Java fn-execution] Drain bundle output before waiting for the ProcessBundleResponse [beam]
Posted by "PabloPauling (via GitHub)" <gi...@apache.org>.
PabloPauling commented on code in PR #40721:
URL: https://github.com/apache/beam/pull/40721#discussion_r4376836165
##########
runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/SdkHarnessClient.java:
##########
@@ -516,22 +516,58 @@ public void close() throws Exception {
} finally {
beamFnDataOutboundAggregator.discard();
}
+ // Consume the bundle's output BEFORE waiting for the ProcessBundleResponse. The inbound
+ // queue is bounded, and the SDK does not finish the bundle until its output has been sent.
+ // Waiting for the response first therefore deadlocks once a single bundle produces more
+ // output than the inbound queue and the SDK's own send buffer hold together: the SDK
+ // blocks sending, the response never arrives, and nothing drains the queue. Batch runners
+ // that hand a whole partition to the SDK as one bundle hit this with ~1 GB of output.
+ boolean inboundDrained = false;
+ Exception drainException = null;
+ if (exception == null && beamFnDataInboundObserver.isPresent()) {
+ BeamFnDataInboundObserver inboundObserver = beamFnDataInboundObserver.get();
+ // Async: when the response is already complete, a synchronous callback would run on
+ // this thread and could block putting elements into the queue this thread must drain.
+ response.whenCompleteAsync(
+ (completedResponse, failure) -> {
+ try {
+ if (failure != null) {
+ // The SDK will not send the rest of the output; unblock the drain below.
+ inboundObserver.close();
+ } else if (completedResponse.hasElements()) {
+ inboundObserver.accept(completedResponse.getElements());
+ }
+ } catch (Exception e) {
+ // Only possible once the drain has failed and cancelled the queue, and that
+ // failure is reported by close().
+ LOG.debug("Dropping the response elements of failed bundle {}", bundleId, e);
+ }
+ });
+ inboundDrained = true;
Review Comment:
Done in 900eb56. Removed inboundDrained and now close the inbound observer unconditionally right after the drain, since close() is idempotent. This also covers the case where sending the input failed, so the close at the end is gone.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org