bgeng777 opened a new issue, #40774:
URL: https://github.com/apache/beam/issues/40774
### What happened?
The Python SDK data channel can replace an already recorded reader failure
with
`RuntimeError("Channel closed prematurely.")`. The resulting bundle error
contains
`_queue.Empty` as context, but omits the actual reader exception.
In `_GrpcDataChannel._read_inputs`, the failure is logged and stored in
`self._exception`, then `self._closed` is set in `finally`. In
`input_elements`,
the closed-channel check runs before the saved-exception check. Consequently,
`SdkHarness._execute` serializes a generic closed-channel traceback into
`InstructionResponse.error` without the recorded failure.
This was observed while investigating PyFlink diagnostics. The following
reproducer uses Beam only and requires no Flink installation.
### Reproducer
With `apache-beam==2.75.0`, save this as `repro.py` and run `python
repro.py`:
```python
import unittest
from apache_beam.runners.worker import data_plane
class ChannelErrorTest(unittest.TestCase):
def test_saved_reader_failure_is_preserved(self):
channel = data_plane._GrpcDataChannel()
root_cause = ValueError("original data-plane failure")
def failed_inputs():
raise root_cause
yield
try:
with self.assertLogs(data_plane._LOGGER, level="ERROR"):
with self.assertRaises(ValueError):
channel._read_inputs(failed_inputs())
self.assertIs(channel._exception, root_cause)
self.assertTrue(channel._closed)
with self.assertRaisesRegex(
RuntimeError, "Channel closed prematurely") as caught:
list(channel.input_elements("instruction", ["input"]))
self.assertIs(caught.exception.__cause__, root_cause)
finally:
channel._exception = None
channel.close()
if __name__ == "__main__":
unittest.main()
```
Actual result:
```text
AssertionError: None is not ValueError('original data-plane failure')
```
Expected: the already recorded reader exception is retained as the explicit
cause of the premature-close error and appears in the bundle error traceback.
### Suggested change
In `_GrpcDataChannel.input_elements`:
```python
if self._closed:
raise RuntimeError(
'Channel closed prematurely.') from self._exception
```
This preserves the existing exception type, message, and abort ordering.
When no exception was recorded, `from None` suppresses the unrelated queue
timeout.
### Verification
Reproduced with Python 3.12.7 and Beam 2.75.0, and with the corresponding
source
files fetched from master. A separate local gRPC test returns
`StatusCode.UNAVAILABLE` with a known detail string, invokes the real
`_read_inputs` and `SdkHarness._execute`, and checks the serialized
`InstructionResponse.error`. The status, detail and original stack are
missing
before this change and present afterward.
Using Beam 2.75.0 dependencies, the three existing data-plane tests plus
seven
regression/integration checks pass after the proposed change (10 total).
The matching master source/tests plus the additional checks pass too (12
total);
this is not a complete master checkout/build.
Related: #20981 reports a data-plane read failure followed by the generic
premature-close error. This issue specifically concerns preserving that saved
failure in the final error response, rather than fixing a transport failure.
### Issue Priority
Priority: 3 (minor)
### Issue Components
Component: Python SDK
AI assistance: OpenAI Codex was used to prepare and validate this report.
--
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]
