PabloPauling opened a new issue, #40720:
URL: https://github.com/apache/beam/issues/40720
### What happened?
Since 2.77.0, a Python stage that produces more than roughly 1 GB of output
in a
single bundle hangs forever on the portable Flink runner in batch mode.
There is
no error and no timeout: the stage stays RUNNING with `numRecordsOut = 0`.
Downstream SDK workers log `Detected input queue delay longer than 300
seconds`.
The same pipeline completes on 2.72.0.
Flink batch hands a stage's whole partition to the SDK as one bundle, so this
hits as soon as a subtask's output reaches about 1 GB. In our case that was a
10M-row read at parallelism 50, about 1.4 GB per subtask.
**Mechanism**
`SdkHarnessClient.ActiveBundle.close()` waits for the `ProcessBundleResponse`
(`MoreFutures.get(response)`) BEFORE it consumes the bundle's output
(`beamFnDataInboundObserver.awaitCompletion()`). Until then, output goes into
`BeamFnDataInboundObserver`'s `CancellableQueue`, which is bounded at 100
messages.
Before #38407, the Python SDK's outbound queue (`data_plane._to_send`) was
unbounded, so the SDK kept accepting output while the runner wasn't draining,
finished the bundle, and sent the response. After #38407 it is a
`ByteLimitedQueue` capped at 100 MB / 10000 messages. Once the runner's 100
messages (~10 MB each) and the SDK's 100 MB are full, the DoFn blocks on
output,
the bundle never finishes, the response never comes, and nothing drains the
queue.
Thread dump of a stuck TaskManager:
```
"CHAIN MapPartition (...)" WAITING on CompletableFuture$Signaller
at org.apache.beam.sdk.util.MoreFutures.get(MoreFutures.java:58)
at
...SdkHarnessClient$BundleProcessor$ActiveBundle.close(SdkHarnessClient.java:522)
at
...DefaultJobBundleFactory$SimpleStageBundleFactory$1.close(DefaultJobBundleFactory.java:563)
at
...FlinkExecutableStageFunction.mapPartition(FlinkExecutableStageFunction.java:269)
"grpc-default-executor-11" WAITING on
AbstractQueuedSynchronizer$ConditionObject
at org.apache.beam.sdk.fn.CancellableQueue.put(CancellableQueue.java:68)
at
org.apache.beam.sdk.fn.data.BeamFnDataInboundObserver.accept(BeamFnDataInboundObserver.java:110)
at
...BeamFnDataGrpcMultiplexer$InboundObserver.forwardToConsumerForInstructionId(...)
```
The Java side is the same on master. Any runner that uses `SdkHarnessClient`
with large bundles is exposed. #40606 addresses a different stall (`onNext`
blocking the shared gRPC executor). It does not change this ordering, so it
does
not fix this.
**Workaround:** set `data_plane._DEFAULT_SEND_QUEUE_MAX_ELEMENTS` and
`_DEFAULT_SEND_QUEUE_MAX_BYTES` to 0 in the worker image, which restores the
2.72 behaviour.
**Issue Priority:** P1 (hangs jobs with no error)
**Components:** runner-flink, sdk-java-core (fn-execution), sdk-py-harness
--
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]
