From 480f5c92afb92af0c69ed3671ea53c5fb7f0a50a Mon Sep 17 00:00:00 2001 From: Valentyn Tymofieiev Date: Wed, 14 Oct 2026 17:12:22 -0700 Subject: [PATCH 1/2] Enable throttler. --- .../throttling/ReactiveThrottler.java | 20 ++++--- .../ml/inference/remote/RemoteInference.java | 6 +- .../inference/remote/RemoteInferenceTest.java | 3 - .../io/components/adaptive_throttler.py | 44 +++++++++----- .../io/components/adaptive_throttler_test.py | 57 +++++++++++++++++++ sdks/python/apache_beam/ml/inference/base.py | 8 ++- .../apache_beam/ml/inference/base_test.py | 2 +- 7 files changed, 110 insertions(+), 30 deletions(-) diff --git a/sdks/java/io/components/src/main/java/org/apache/beam/sdk/io/components/throttling/ReactiveThrottler.java b/sdks/java/io/components/src/main/java/org/apache/beam/sdk/io/components/throttling/ReactiveThrottler.java index 65ab23e3defe..bf603cd1454e 100644 --- a/sdks/java/io/components/src/main/java/org/apache/beam/sdk/io/components/throttling/ReactiveThrottler.java +++ b/sdks/java/io/components/src/main/java/org/apache/beam/sdk/io/components/throttling/ReactiveThrottler.java @@ -28,9 +28,12 @@ *

For usage, instantiate one instance of a ReactiveThrottler class for a PTransform. When making * remote calls to a service, preface that call with the throttle() method to potentially * pre-emptively throttle the request. This will throttle future calls based on the failure rate of - * preceding calls, with higher failure rates leading to longer periods of throttling to allow - * system recovery. capture the timestamp of the attempted request, then execute the request code. - * On a success, call successfulRequest(timestamp) to report the success to the throttler. + * preceding calls, with higher failure rates making it more likely that a call is delayed, to allow + * system recovery. A throttled call is delayed once by throttleDelaySecs and then proceeds; it is + * never dropped. Call throttle() exactly once per attempt (including retries), since each call is + * recorded as one request attempt. Then capture the timestamp of the attempted request and execute + * the request code. On a success, call successfulRequest(timestamp) to report the success to the + * throttler; attempts that are not reported as successful count as failures. */ @SuppressFBWarnings( value = "CT_CONSTRUCTOR_THROW", @@ -70,10 +73,13 @@ public ReactiveThrottler( } /** - * Stops request code from advancing while the underlying AdaptiveThrottler is signaling to - * preemptively throttle the request. Automatically handles logging the throttling and signaling - * to the SDK harness that the request is being throttled. This should be called in any context - * where a call to a remote service is being contacted prior to the call being performed. + * Delays the request by throttleDelaySecs if the underlying AdaptiveThrottler elects to + * preemptively throttle it, based on the failure rate of recent requests. Automatically handles + * logging the throttling and signaling to the SDK harness that the request is being throttled. + * + *

This should be called once for each attempt to contact a remote service (including + * retries), prior to the call being performed. Each call is recorded as one request attempt, and + * delays the request at most once. */ public void throttle() throws InterruptedException { if (throttleRequest(System.currentTimeMillis())) { diff --git a/sdks/java/ml/inference/remote/src/main/java/org/apache/beam/sdk/ml/inference/remote/RemoteInference.java b/sdks/java/ml/inference/remote/src/main/java/org/apache/beam/sdk/ml/inference/remote/RemoteInference.java index a998f4d0f077..ee464ceb78f7 100644 --- a/sdks/java/ml/inference/remote/src/main/java/org/apache/beam/sdk/ml/inference/remote/RemoteInference.java +++ b/sdks/java/ml/inference/remote/src/main/java/org/apache/beam/sdk/ml/inference/remote/RemoteInference.java @@ -143,7 +143,9 @@ public Invoke withThrottleDelaySecs(int throttleDelaySecs) { /** * Configures the length of history to consider when setting throttling probability. Defaults to - * a sample period of 1000ms. For more context, see {@link AdaptiveThrottler} + * a sample period of 120000ms. Should be much larger than the throttle delay and the retry + * backoff, otherwise failures expire from the history before they can influence throttling. For + * more context, see {@link AdaptiveThrottler} */ public Invoke withSamplePeriodMs(long samplePeriodMs) { checkArgument(samplePeriodMs > 0, "samplePeriodMs must be positive"); @@ -239,7 +241,7 @@ static class RemoteInferenceFninvoke() .handler(MockThrottlingHandler.class) .withBatchConfig(batchConfig) - // Use large sample periods so the 1s retry delay doesn't flush the history - .withSamplePeriodMs(60000L) - .withSampleUpdateMs(60000L) // Set to 1 second to minimize test wait time while still verifying throttling .withThrottleDelaySecs(1) .withOverloadRatio(1.1) diff --git a/sdks/python/apache_beam/io/components/adaptive_throttler.py b/sdks/python/apache_beam/io/components/adaptive_throttler.py index 3c22891ee8a3..4df5da5a3902 100644 --- a/sdks/python/apache_beam/io/components/adaptive_throttler.py +++ b/sdks/python/apache_beam/io/components/adaptive_throttler.py @@ -98,6 +98,11 @@ def throttle_request(self, now): it returns true, drop or delay that request (calling this function again after the delay). + Every call is recorded as one request attempt, including calls that return + True: requests rejected locally by the client still + count towards "requests". A caller that re-checks a delayed request + therefore records an additional attempt on every check. + Args: now: int, time in ms since the epoch Returns: @@ -127,16 +132,21 @@ class ReactiveThrottler(AdaptiveThrottler): PTransform. When making remote calls to a service, preface that call with the throttle() method to potentially pre-emptively throttle the request. This will throttle future calls based on the failure rate of preceding calls, - with higher failure rates leading to longer periods of throttling to allow - system recovery. capture the timestamp of the attempted request, then execute - the request code. On a success, call successful_request(timestamp) to report - the success to the throttler. This flow looks like the following: + with higher failure rates making it more likely that a call is delayed, to + allow system recovery. A throttled call is delayed once by + throttle_delay_secs and then proceeds; it is never dropped. Call throttle() + once per attempt (including retries), since each call is recorded as + one request attempt. Then capture the timestamp of the attempted request (in + milliseconds since the epoch) and execute the request code. On a success, + call successful_request(timestamp) to report the success to the throttler; + attempts that are not reported as successful count as failures. This flow + looks like the following: def remote_call(): throttler.throttle() try: - timestamp = time.time() + timestamp = time.time() * 1000 result = make_request() throttler.successful_request(timestamp) return result @@ -155,7 +165,9 @@ def __init__( Args: window_ms: int, length of history to consider, in ms, to set - throttling. + throttling. Should be much larger than throttle_delay_secs and any + retry backoff between attempts, otherwise failures expire from the + history before they can influence throttling. bucket_ms: int, granularity of time buckets that we store data in, in ms. overload_ratio: float, the target ratio between requests sent and @@ -163,8 +175,8 @@ def __init__( https://landing.google.com/sre/book/chapters/handling-overload.html. namespace: str, the namespace to use for logging and signaling throttling is occurring - throttle_delay_secs: int, the amount of time in seconds to wait - after preemptively throttled requests + throttle_delay_secs: int, the amount of time in seconds to delay a + request that is preemptively throttled """ self.throttling_signaler = ThrottlingSignaler(namespace=namespace) self.logger = logging.getLogger(namespace) @@ -173,14 +185,16 @@ def __init__( window_ms=window_ms, bucket_ms=bucket_ms, overload_ratio=overload_ratio) def throttle(self): - """ Stops request code from advancing while the underlying - AdaptiveThrottler is signaling to preemptively throttle the request. - Automatically handles logging the throttling and signaling to the SDK - harness that the request is being throttled. This should be called in any - context where a call to a remote service is being contacted prior to the - call being performed. + """ Delays the request by throttle_delay_secs if the underlying + AdaptiveThrottler elects to preemptively throttle it, based on the failure + rate of recent requests. Automatically handles logging the throttling and + signaling to the SDK harness that the request is being throttled. + + This should be called once for each attempt to contact a remote + service (including retries), prior to the call being performed. Each call + is recorded as one request attempt, and delays the request at most once. """ - while self.throttle_request(time.time() * _SECONDS_TO_MILLISECONDS): + if self.throttle_request(time.time() * _SECONDS_TO_MILLISECONDS): self.logger.info( "Delaying request for %d seconds due to previous failures", self.throttle_delay_secs) diff --git a/sdks/python/apache_beam/io/components/adaptive_throttler_test.py b/sdks/python/apache_beam/io/components/adaptive_throttler_test.py index d78908486780..10df3dc85e78 100644 --- a/sdks/python/apache_beam/io/components/adaptive_throttler_test.py +++ b/sdks/python/apache_beam/io/components/adaptive_throttler_test.py @@ -19,9 +19,11 @@ import unittest +from mock import MagicMock from mock import patch from apache_beam.io.components.adaptive_throttler import AdaptiveThrottler +from apache_beam.io.components.adaptive_throttler import ReactiveThrottler class AdaptiveThrottlerTest(unittest.TestCase): @@ -105,5 +107,60 @@ def test_throttling_after_errors(self, mock_random): self.assertEqual(t < AdaptiveThrottlerTest.START_TIME + 14, throttled) +class ReactiveThrottlerTest(unittest.TestCase): + + START_TIME_SECS = 1500000000 + THROTTLE_DELAY_SECS = 5 + + def setUp(self): + self._throttler = ReactiveThrottler( + window_ms=60000, + bucket_ms=1000, + overload_ratio=2, + namespace='ReactiveThrottlerTest', + throttle_delay_secs=ReactiveThrottlerTest.THROTTLE_DELAY_SECS) + self._mock_signaler = MagicMock() + self._throttler.throttling_signaler = self._mock_signaler + sleep_patcher = patch( + 'apache_beam.io.components.adaptive_throttler.time.sleep') + self._mock_sleep = sleep_patcher.start() + self.addCleanup(sleep_patcher.stop) + + def test_throttle_does_not_delay_unthrottled_request(self): + with patch.object(self._throttler, 'throttle_request', return_value=False): + self._throttler.throttle() + self._mock_sleep.assert_not_called() + self._mock_signaler.signal_throttled.assert_not_called() + + def test_throttle_delays_throttled_request_once(self): + with patch.object(self._throttler, + 'throttle_request', + side_effect=[True, False]) as mock_throttle_request: + self._throttler.throttle() + mock_throttle_request.assert_called_once() + self._mock_sleep.assert_called_once_with( + ReactiveThrottlerTest.THROTTLE_DELAY_SECS) + self._mock_signaler.signal_throttled.assert_called_once_with( + ReactiveThrottlerTest.THROTTLE_DELAY_SECS) + + def test_throttle_records_one_request_per_call(self): + # Throttle whenever the throttling probability is positive. The finite + # side_effect makes an implementation that re-checks after the delay fail + # fast instead of looping forever. + self._throttler._random = MagicMock() + self._throttler._random.uniform.side_effect = [0.0] * 3 + with patch('apache_beam.io.components.adaptive_throttler.time.time', + return_value=ReactiveThrottlerTest.START_TIME_SECS): + # None of the requests succeed, i.e. successful_request() is not called. + for _ in range(3): + self._throttler.throttle() + + now_ms = ReactiveThrottlerTest.START_TIME_SECS * 1000 + self.assertEqual(3, self._throttler._all_requests.sum(now_ms)) + self.assertEqual(0.75, self._throttler._throttling_probability(now_ms)) + # The first request has no history to be throttled on. + self.assertEqual(2, self._mock_sleep.call_count) + + if __name__ == '__main__': unittest.main() diff --git a/sdks/python/apache_beam/ml/inference/base.py b/sdks/python/apache_beam/ml/inference/base.py index 8c4b10f8573d..26bab98d3fd1 100644 --- a/sdks/python/apache_beam/ml/inference/base.py +++ b/sdks/python/apache_beam/ml/inference/base.py @@ -414,7 +414,7 @@ def __init__( throttle_delay_secs: int = 5, retry_filter: Callable[[Exception], bool] = lambda x: True, *, - window_ms: int = 1 * _MILLISECOND_TO_SECOND, + window_ms: int = 120 * _MILLISECOND_TO_SECOND, bucket_ms: int = 1 * _MILLISECOND_TO_SECOND, overload_ratio: float = 2, rate_limiter: Optional[RateLimiter] = None): @@ -431,11 +431,15 @@ def __init__( num_retries: the maximum number of times to retry a request on retriable errors before failing throttle_delay_secs: the amount of time to throttle when the client-side - elects to throttle + elects to throttle. A throttled request is delayed once by this amount + and then sent. retry_filter: a function accepting an exception as an argument and returning a boolean. On a true return, the run_inference call will be retried. Defaults to always retrying. window_ms: length of history to consider, in ms, to set throttling. + Should be much larger than throttle_delay_secs and the retry backoff, + otherwise failures expire from the history before they can influence + throttling. bucket_ms: granularity of time buckets that we store data in, in ms. overload_ratio: the target ratio between requests sent and successful requests. This is "K" in the formula in diff --git a/sdks/python/apache_beam/ml/inference/base_test.py b/sdks/python/apache_beam/ml/inference/base_test.py index 8cf408db76f4..7715070f281a 100644 --- a/sdks/python/apache_beam/ml/inference/base_test.py +++ b/sdks/python/apache_beam/ml/inference/base_test.py @@ -1989,7 +1989,7 @@ def __init__( super().__init__( namespace='FakeRemoteModelHandler', retry_filter=retry_filter, - num_retries=2, + num_retries=1, throttle_delay_secs=1) def create_client(self): From 5590cee1d53faf5e6c304188b6dd18fb662f4071 Mon Sep 17 00:00:00 2001 From: Valentyn Tymofieiev Date: Thu, 15 Oct 2026 16:43:57 -0700 Subject: [PATCH 2/2] Preserve while loop instead of if. --- .../throttling/ReactiveThrottler.java | 20 +++---- .../ml/inference/remote/RemoteInference.java | 4 +- .../io/components/adaptive_throttler.py | 44 +++++--------- .../io/components/adaptive_throttler_test.py | 57 ------------------- sdks/python/apache_beam/ml/inference/base.py | 6 +- .../apache_beam/ml/inference/base_test.py | 2 +- 6 files changed, 25 insertions(+), 108 deletions(-) diff --git a/sdks/java/io/components/src/main/java/org/apache/beam/sdk/io/components/throttling/ReactiveThrottler.java b/sdks/java/io/components/src/main/java/org/apache/beam/sdk/io/components/throttling/ReactiveThrottler.java index bf603cd1454e..65ab23e3defe 100644 --- a/sdks/java/io/components/src/main/java/org/apache/beam/sdk/io/components/throttling/ReactiveThrottler.java +++ b/sdks/java/io/components/src/main/java/org/apache/beam/sdk/io/components/throttling/ReactiveThrottler.java @@ -28,12 +28,9 @@ *

For usage, instantiate one instance of a ReactiveThrottler class for a PTransform. When making * remote calls to a service, preface that call with the throttle() method to potentially * pre-emptively throttle the request. This will throttle future calls based on the failure rate of - * preceding calls, with higher failure rates making it more likely that a call is delayed, to allow - * system recovery. A throttled call is delayed once by throttleDelaySecs and then proceeds; it is - * never dropped. Call throttle() exactly once per attempt (including retries), since each call is - * recorded as one request attempt. Then capture the timestamp of the attempted request and execute - * the request code. On a success, call successfulRequest(timestamp) to report the success to the - * throttler; attempts that are not reported as successful count as failures. + * preceding calls, with higher failure rates leading to longer periods of throttling to allow + * system recovery. capture the timestamp of the attempted request, then execute the request code. + * On a success, call successfulRequest(timestamp) to report the success to the throttler. */ @SuppressFBWarnings( value = "CT_CONSTRUCTOR_THROW", @@ -73,13 +70,10 @@ public ReactiveThrottler( } /** - * Delays the request by throttleDelaySecs if the underlying AdaptiveThrottler elects to - * preemptively throttle it, based on the failure rate of recent requests. Automatically handles - * logging the throttling and signaling to the SDK harness that the request is being throttled. - * - *

This should be called once for each attempt to contact a remote service (including - * retries), prior to the call being performed. Each call is recorded as one request attempt, and - * delays the request at most once. + * Stops request code from advancing while the underlying AdaptiveThrottler is signaling to + * preemptively throttle the request. Automatically handles logging the throttling and signaling + * to the SDK harness that the request is being throttled. This should be called in any context + * where a call to a remote service is being contacted prior to the call being performed. */ public void throttle() throws InterruptedException { if (throttleRequest(System.currentTimeMillis())) { diff --git a/sdks/java/ml/inference/remote/src/main/java/org/apache/beam/sdk/ml/inference/remote/RemoteInference.java b/sdks/java/ml/inference/remote/src/main/java/org/apache/beam/sdk/ml/inference/remote/RemoteInference.java index ee464ceb78f7..3dd4158069b0 100644 --- a/sdks/java/ml/inference/remote/src/main/java/org/apache/beam/sdk/ml/inference/remote/RemoteInference.java +++ b/sdks/java/ml/inference/remote/src/main/java/org/apache/beam/sdk/ml/inference/remote/RemoteInference.java @@ -143,9 +143,7 @@ public Invoke withThrottleDelaySecs(int throttleDelaySecs) { /** * Configures the length of history to consider when setting throttling probability. Defaults to - * a sample period of 120000ms. Should be much larger than the throttle delay and the retry - * backoff, otherwise failures expire from the history before they can influence throttling. For - * more context, see {@link AdaptiveThrottler} + * a sample period of 120000ms. For more context, see {@link AdaptiveThrottler} */ public Invoke withSamplePeriodMs(long samplePeriodMs) { checkArgument(samplePeriodMs > 0, "samplePeriodMs must be positive"); diff --git a/sdks/python/apache_beam/io/components/adaptive_throttler.py b/sdks/python/apache_beam/io/components/adaptive_throttler.py index 4df5da5a3902..06f2d83d7142 100644 --- a/sdks/python/apache_beam/io/components/adaptive_throttler.py +++ b/sdks/python/apache_beam/io/components/adaptive_throttler.py @@ -98,11 +98,6 @@ def throttle_request(self, now): it returns true, drop or delay that request (calling this function again after the delay). - Every call is recorded as one request attempt, including calls that return - True: requests rejected locally by the client still - count towards "requests". A caller that re-checks a delayed request - therefore records an additional attempt on every check. - Args: now: int, time in ms since the epoch Returns: @@ -132,16 +127,11 @@ class ReactiveThrottler(AdaptiveThrottler): PTransform. When making remote calls to a service, preface that call with the throttle() method to potentially pre-emptively throttle the request. This will throttle future calls based on the failure rate of preceding calls, - with higher failure rates making it more likely that a call is delayed, to - allow system recovery. A throttled call is delayed once by - throttle_delay_secs and then proceeds; it is never dropped. Call throttle() - once per attempt (including retries), since each call is recorded as - one request attempt. Then capture the timestamp of the attempted request (in - milliseconds since the epoch) and execute the request code. On a success, - call successful_request(timestamp) to report the success to the throttler; - attempts that are not reported as successful count as failures. This flow - looks like the following: - + with higher failure rates leading to longer periods of throttling to allow + system recovery. capture the timestamp of the attempted request, then execute + the request code. On a success, call successful_request(timestamp) to report + the success to the throttler. This flow looks like the following: + def remote_call(): throttler.throttle() @@ -165,9 +155,7 @@ def __init__( Args: window_ms: int, length of history to consider, in ms, to set - throttling. Should be much larger than throttle_delay_secs and any - retry backoff between attempts, otherwise failures expire from the - history before they can influence throttling. + throttling. bucket_ms: int, granularity of time buckets that we store data in, in ms. overload_ratio: float, the target ratio between requests sent and @@ -175,8 +163,8 @@ def __init__( https://landing.google.com/sre/book/chapters/handling-overload.html. namespace: str, the namespace to use for logging and signaling throttling is occurring - throttle_delay_secs: int, the amount of time in seconds to delay a - request that is preemptively throttled + throttle_delay_secs: int, the amount of time in seconds to wait + after preemptively throttled requests """ self.throttling_signaler = ThrottlingSignaler(namespace=namespace) self.logger = logging.getLogger(namespace) @@ -185,16 +173,14 @@ def __init__( window_ms=window_ms, bucket_ms=bucket_ms, overload_ratio=overload_ratio) def throttle(self): - """ Delays the request by throttle_delay_secs if the underlying - AdaptiveThrottler elects to preemptively throttle it, based on the failure - rate of recent requests. Automatically handles logging the throttling and - signaling to the SDK harness that the request is being throttled. - - This should be called once for each attempt to contact a remote - service (including retries), prior to the call being performed. Each call - is recorded as one request attempt, and delays the request at most once. + """ Stops request code from advancing while the underlying + AdaptiveThrottler is signaling to preemptively throttle the request. + Automatically handles logging the throttling and signaling to the SDK + harness that the request is being throttled. This should be called in any + context where a call to a remote service is being contacted prior to the + call being performed. """ - if self.throttle_request(time.time() * _SECONDS_TO_MILLISECONDS): + while self.throttle_request(time.time() * _SECONDS_TO_MILLISECONDS): self.logger.info( "Delaying request for %d seconds due to previous failures", self.throttle_delay_secs) diff --git a/sdks/python/apache_beam/io/components/adaptive_throttler_test.py b/sdks/python/apache_beam/io/components/adaptive_throttler_test.py index 10df3dc85e78..d78908486780 100644 --- a/sdks/python/apache_beam/io/components/adaptive_throttler_test.py +++ b/sdks/python/apache_beam/io/components/adaptive_throttler_test.py @@ -19,11 +19,9 @@ import unittest -from mock import MagicMock from mock import patch from apache_beam.io.components.adaptive_throttler import AdaptiveThrottler -from apache_beam.io.components.adaptive_throttler import ReactiveThrottler class AdaptiveThrottlerTest(unittest.TestCase): @@ -107,60 +105,5 @@ def test_throttling_after_errors(self, mock_random): self.assertEqual(t < AdaptiveThrottlerTest.START_TIME + 14, throttled) -class ReactiveThrottlerTest(unittest.TestCase): - - START_TIME_SECS = 1500000000 - THROTTLE_DELAY_SECS = 5 - - def setUp(self): - self._throttler = ReactiveThrottler( - window_ms=60000, - bucket_ms=1000, - overload_ratio=2, - namespace='ReactiveThrottlerTest', - throttle_delay_secs=ReactiveThrottlerTest.THROTTLE_DELAY_SECS) - self._mock_signaler = MagicMock() - self._throttler.throttling_signaler = self._mock_signaler - sleep_patcher = patch( - 'apache_beam.io.components.adaptive_throttler.time.sleep') - self._mock_sleep = sleep_patcher.start() - self.addCleanup(sleep_patcher.stop) - - def test_throttle_does_not_delay_unthrottled_request(self): - with patch.object(self._throttler, 'throttle_request', return_value=False): - self._throttler.throttle() - self._mock_sleep.assert_not_called() - self._mock_signaler.signal_throttled.assert_not_called() - - def test_throttle_delays_throttled_request_once(self): - with patch.object(self._throttler, - 'throttle_request', - side_effect=[True, False]) as mock_throttle_request: - self._throttler.throttle() - mock_throttle_request.assert_called_once() - self._mock_sleep.assert_called_once_with( - ReactiveThrottlerTest.THROTTLE_DELAY_SECS) - self._mock_signaler.signal_throttled.assert_called_once_with( - ReactiveThrottlerTest.THROTTLE_DELAY_SECS) - - def test_throttle_records_one_request_per_call(self): - # Throttle whenever the throttling probability is positive. The finite - # side_effect makes an implementation that re-checks after the delay fail - # fast instead of looping forever. - self._throttler._random = MagicMock() - self._throttler._random.uniform.side_effect = [0.0] * 3 - with patch('apache_beam.io.components.adaptive_throttler.time.time', - return_value=ReactiveThrottlerTest.START_TIME_SECS): - # None of the requests succeed, i.e. successful_request() is not called. - for _ in range(3): - self._throttler.throttle() - - now_ms = ReactiveThrottlerTest.START_TIME_SECS * 1000 - self.assertEqual(3, self._throttler._all_requests.sum(now_ms)) - self.assertEqual(0.75, self._throttler._throttling_probability(now_ms)) - # The first request has no history to be throttled on. - self.assertEqual(2, self._mock_sleep.call_count) - - if __name__ == '__main__': unittest.main() diff --git a/sdks/python/apache_beam/ml/inference/base.py b/sdks/python/apache_beam/ml/inference/base.py index 26bab98d3fd1..f923555e7845 100644 --- a/sdks/python/apache_beam/ml/inference/base.py +++ b/sdks/python/apache_beam/ml/inference/base.py @@ -431,15 +431,11 @@ def __init__( num_retries: the maximum number of times to retry a request on retriable errors before failing throttle_delay_secs: the amount of time to throttle when the client-side - elects to throttle. A throttled request is delayed once by this amount - and then sent. + elects to throttle retry_filter: a function accepting an exception as an argument and returning a boolean. On a true return, the run_inference call will be retried. Defaults to always retrying. window_ms: length of history to consider, in ms, to set throttling. - Should be much larger than throttle_delay_secs and the retry backoff, - otherwise failures expire from the history before they can influence - throttling. bucket_ms: granularity of time buckets that we store data in, in ms. overload_ratio: the target ratio between requests sent and successful requests. This is "K" in the formula in diff --git a/sdks/python/apache_beam/ml/inference/base_test.py b/sdks/python/apache_beam/ml/inference/base_test.py index 7715070f281a..8cf408db76f4 100644 --- a/sdks/python/apache_beam/ml/inference/base_test.py +++ b/sdks/python/apache_beam/ml/inference/base_test.py @@ -1989,7 +1989,7 @@ def __init__( super().__init__( namespace='FakeRemoteModelHandler', retry_filter=retry_filter, - num_retries=1, + num_retries=2, throttle_delay_secs=1) def create_client(self):