Remove recursion

Author: Quinn-With-Two-Ns

temporal-sdk/src/main/java/io/temporal/worker/tuning/ResourceBasedSlotSupplier.java

@@ -4,6 +4,7 @@
 import java.time.Instant;
 import java.util.Optional;
 import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicReference;
 
 /** Implements a {@link SlotSupplier} based on resource usage for a particular slot type. */
 public class ResourceBasedSlotSupplier<SI extends SlotInfo> implements SlotSupplier<SI> {
@@ -190,35 +191,62 @@ public SlotSupplierFuture reserveSlot(SlotReserveContext<SI> ctx) throws Excepti
   }
 
   private SlotSupplierFuture scheduleSlotAcquisition(SlotReserveContext<SI> ctx) {
+    CompletableFuture<SlotPermit> resultFuture = new CompletableFuture<>();
+    AtomicReference<ScheduledFuture<?>> taskRef = new AtomicReference<>();
+
+    Runnable pollingTask =
+        new Runnable() {
+          @Override
+          public void run() {
+            if (resultFuture.isDone()) {
+              return; // Already completed or cancelled
+            }
+
+            try {
+              Optional<SlotPermit> permit = tryReserveSlot(ctx);
+              if (permit.isPresent()) {
+                resultFuture.complete(permit.get());
+              } else {
+                // Calculate delay respecting ramp throttle on each retry
+                Duration mustWaitFor;
+                try {
+                  mustWaitFor = options.getRampThrottle().minus(timeSinceLastSlotIssued());
+                } catch (ArithmeticException e) {
+                  mustWaitFor = Duration.ZERO;
+                }
+
+                // Use at least 10ms to avoid tight spinning, but respect ramp throttle if longer
+                long delayMs = Math.max(10, mustWaitFor.toMillis());
+                taskRef.set(scheduler.schedule(this, delayMs, TimeUnit.MILLISECONDS));
+              }
+            } catch (Exception e) {
+              resultFuture.completeExceptionally(e);
+            }
+          }
+        };
+
+    // Calculate initial delay based on ramp throttle
     Duration mustWaitFor;
     try {
       mustWaitFor = options.getRampThrottle().minus(timeSinceLastSlotIssued());
     } catch (ArithmeticException e) {
       mustWaitFor = Duration.ZERO;
     }
 
-    CompletableFuture<Void> permitFuture;
-    if (mustWaitFor.compareTo(Duration.ZERO) > 0) {
-      permitFuture =
-          CompletableFuture.supplyAsync(() -> null, delayedExecutor(mustWaitFor.toMillis()));
-    } else {
-      permitFuture = CompletableFuture.completedFuture(null);
-    }
+    long initialDelayMs = Math.max(0, mustWaitFor.toMillis());
+
+    // Schedule the initial attempt
+    taskRef.set(scheduler.schedule(pollingTask, initialDelayMs, TimeUnit.MILLISECONDS));
 
-    // After the delay, try to reserve the slot
     return SlotSupplierFuture.fromCompletableFuture(
-        permitFuture.thenCompose(
-            ignored -> {
-              Optional<SlotPermit> permit = tryReserveSlot(ctx);
-              // If we couldn't get a slot this time, delay for a short period and try again
-              return permit
-                  .map(CompletableFuture::completedFuture)
-                  .orElseGet(
-                      () ->
-                          CompletableFuture.supplyAsync(() -> null, delayedExecutor(10))
-                              .thenCompose(ig -> scheduleSlotAcquisition(ctx)));
-            }),
-        () -> permitFuture.cancel(true));
+        resultFuture,
+        () -> {
+          // Cancel the scheduled task when aborting
+          ScheduledFuture<?> task = taskRef.get();
+          if (task != null) {
+            task.cancel(true);
+          }
+        });
   }
 
   @Override
@@ -248,11 +276,6 @@ private Duration timeSinceLastSlotIssued() {
     return Duration.between(lastSlotIssuedAt, Instant.now());
   }
 
-  // Polyfill for Java 9 delayedExecutor
-  private Executor delayedExecutor(long delay) {
-    return r -> scheduler.schedule(() -> scheduler.execute(r), delay, TimeUnit.MILLISECONDS);
-  }
-
   private static ScheduledExecutorService getDefaultScheduler() {
     synchronized (ResourceBasedSlotSupplier.class) {
       if (defaultScheduler == null) {