This is an automated email from the ASF dual-hosted git repository.
Abacn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 4d975e5ec60 Avoid RequestResponseIO throttler access when throttler is
absent (#39112)
4d975e5ec60 is described below
commit 4d975e5ec608b3db65ce826e2141fbe2f1aebc19
Author: Minh Vu <[email protected]>
AuthorDate: Mon Jul 6 21:59:18 2026 +0200
Avoid RequestResponseIO throttler access when throttler is absent (#39112)
* Avoid throttler access when RequestResponseIO has no throttler
* Use explicit None checks for throttler in RequestResponseIO
* sync PR head for explicit None checks
* Tighten CI-flaky tests around async teardown and minibatch size
* Fix formatter regression in pytorch batch size assertion
* Keep RequestResponseIO PR focused
---
sdks/python/apache_beam/io/requestresponse.py | 5 +++--
sdks/python/apache_beam/io/requestresponse_test.py | 10 ++++++++++
2 files changed, 13 insertions(+), 2 deletions(-)
diff --git a/sdks/python/apache_beam/io/requestresponse.py
b/sdks/python/apache_beam/io/requestresponse.py
index 9fdf33e2299..ef570f4d855 100644
--- a/sdks/python/apache_beam/io/requestresponse.py
+++ b/sdks/python/apache_beam/io/requestresponse.py
@@ -358,7 +358,7 @@ class _CallDoFn(beam.DoFn):
self._metrics_collector.requests.inc(1)
is_throttled_request = False
- if self._throttler:
+ if self._throttler is not None:
while self._throttler.throttler.throttle_request(time.time() *
MSEC_TO_SEC):
_LOGGER.info(
@@ -375,7 +375,8 @@ class _CallDoFn(beam.DoFn):
response = self._repeater.repeat(
self._caller, request, self._timeout, self._metrics_collector)
self._metrics_collector.responses.inc(1)
- self._throttler.throttler.successful_request(req_time * MSEC_TO_SEC)
+ if self._throttler is not None:
+ self._throttler.throttler.successful_request(req_time * MSEC_TO_SEC)
yield response
except Exception as e:
raise e
diff --git a/sdks/python/apache_beam/io/requestresponse_test.py
b/sdks/python/apache_beam/io/requestresponse_test.py
index f88df9657da..7d4d6b4caf3 100644
--- a/sdks/python/apache_beam/io/requestresponse_test.py
+++ b/sdks/python/apache_beam/io/requestresponse_test.py
@@ -95,6 +95,16 @@ class TestCaller(unittest.TestCase):
self.assertIsNotNone(output)
+ def test_valid_call_without_throttler(self):
+ caller = AckCaller()
+ with TestPipeline() as test_pipeline:
+ output = (
+ test_pipeline
+ | beam.Create(["sample_request"])
+ | RequestResponseIO(caller=caller, throttler=None))
+
+ self.assertIsNotNone(output)
+
def test_call_timeout(self):
caller = CallerWithTimeout()
with self.assertRaisesRegex(Exception, "Timeout"):
