[Bug]: Portable Java runners deadlock when one bundle's output exceeds ~1 GB (regression in 2.77 from #38407) #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