This is an automated email from the ASF dual-hosted git repository.
tvalentyn 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 d5f6472f73a Enable throttling logic in RemoteInference. (#40598)
d5f6472f73a is described below
commit d5f6472f73a42cca0bd0bd19ffd17498fdb5f4bb
Author: tvalentyn <[email protected]>
AuthorDate: Fri Oct 16 10:02:41 2026 -0700
Enable throttling logic in RemoteInference. (#40598)
* Enable throttler.
* Preserve while loop instead of if.
---
.../java/org/apache/beam/sdk/ml/inference/remote/RemoteInference.java | 4 ++--
.../org/apache/beam/sdk/ml/inference/remote/RemoteInferenceTest.java | 3 ---
sdks/python/apache_beam/io/components/adaptive_throttler.py | 4 ++--
sdks/python/apache_beam/ml/inference/base.py | 2 +-
4 files changed, 5 insertions(+), 8 deletions(-)
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 a998f4d0f07..3dd4158069b 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,7 @@ public class RemoteInference {
/**
* 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. For more context, see {@link
AdaptiveThrottler}
*/
public Invoke<InputT, OutputT> withSamplePeriodMs(long samplePeriodMs) {
checkArgument(samplePeriodMs > 0, "samplePeriodMs must be positive");
@@ -239,7 +239,7 @@ public class RemoteInference {
this.handlerClass = spec.handler();
this.parameters = spec.parameters();
this.throttleDelaySecs = spec.throttleDelaySecs() != null ?
spec.throttleDelaySecs() : 5;
- this.samplePeriodMs = spec.samplePeriodMs() != null ?
spec.samplePeriodMs() : 1000L;
+ this.samplePeriodMs = spec.samplePeriodMs() != null ?
spec.samplePeriodMs() : 120000L;
this.sampleUpdateMs = spec.sampleUpdateMs() != null ?
spec.sampleUpdateMs() : 1000L;
this.overloadRatio = spec.overloadRatio() != null ?
spec.overloadRatio() : 2.0;
retryHandler = RetryHandler.withDefaults();
diff --git
a/sdks/java/ml/inference/remote/src/test/java/org/apache/beam/sdk/ml/inference/remote/RemoteInferenceTest.java
b/sdks/java/ml/inference/remote/src/test/java/org/apache/beam/sdk/ml/inference/remote/RemoteInferenceTest.java
index 5fe8ea1cce1..0ad37adc146 100644
---
a/sdks/java/ml/inference/remote/src/test/java/org/apache/beam/sdk/ml/inference/remote/RemoteInferenceTest.java
+++
b/sdks/java/ml/inference/remote/src/test/java/org/apache/beam/sdk/ml/inference/remote/RemoteInferenceTest.java
@@ -670,9 +670,6 @@ public class RemoteInferenceTest {
RemoteInference.<TestInput, TestOutput>invoke()
.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 3c22891ee8a..06f2d83d714 100644
--- a/sdks/python/apache_beam/io/components/adaptive_throttler.py
+++ b/sdks/python/apache_beam/io/components/adaptive_throttler.py
@@ -131,12 +131,12 @@ class ReactiveThrottler(AdaptiveThrottler):
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()
try:
- timestamp = time.time()
+ timestamp = time.time() * 1000
result = make_request()
throttler.successful_request(timestamp)
return result
diff --git a/sdks/python/apache_beam/ml/inference/base.py
b/sdks/python/apache_beam/ml/inference/base.py
index 85a287ad752..823a4a3fa71 100644
--- a/sdks/python/apache_beam/ml/inference/base.py
+++ b/sdks/python/apache_beam/ml/inference/base.py
@@ -425,7 +425,7 @@ class RemoteModelHandler(ABC, ModelHandler[ExampleT,
PredictionT, ModelT]):
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):
