Enable throttling logic in RemoteInference.

sdks/java/ml/inference/remote/src/main/java/org/apache/beam/sdk/ml/inference/remote/RemoteInference.java

@@ -143,7 +143,7 @@ public Invoke<InputT, OutputT> 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. For more context, see {@link AdaptiveThrottler}
      */
     public Invoke<InputT, OutputT> withSamplePeriodMs(long samplePeriodMs) {
       checkArgument(samplePeriodMs > 0, "samplePeriodMs must be positive");
@@ -239,7 +239,7 @@ static class RemoteInferenceFn<InputT extends BaseInput, OutputT extends BaseRes
         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();

sdks/java/ml/inference/remote/src/test/java/org/apache/beam/sdk/ml/inference/remote/RemoteInferenceTest.java

@@ -670,9 +670,6 @@ public void testThrottlingBehavior() {
             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)

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

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):