Make eager activity reservation limit configurable

temporal-sdk/src/main/java/io/temporal/internal/worker/EagerActivitySlotsReservation.java

@@ -7,7 +7,6 @@
 import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponse;
 import io.temporal.api.workflowservice.v1.RespondWorkflowTaskCompletedRequest;
 import io.temporal.api.workflowservice.v1.RespondWorkflowTaskCompletedResponse;
-import io.temporal.internal.Config;
 import io.temporal.worker.tuning.SlotPermit;
 import java.io.Closeable;
 import java.util.ArrayList;
@@ -19,10 +18,13 @@
 @NotThreadSafe
 class EagerActivitySlotsReservation implements Closeable {
   private final EagerActivityDispatcher eagerActivityDispatcher;
+  private final int maxReservations;
   private final List<SlotPermit> reservedSlots = new ArrayList<>();
 
-  EagerActivitySlotsReservation(EagerActivityDispatcher eagerActivityDispatcher) {
+  EagerActivitySlotsReservation(
+      EagerActivityDispatcher eagerActivityDispatcher, int maxReservations) {
     this.eagerActivityDispatcher = eagerActivityDispatcher;
+    this.maxReservations = maxReservations;
   }
 
   public void applyToRequest(RespondWorkflowTaskCompletedRequest.Builder mutableRequest) {
@@ -33,7 +35,7 @@ public void applyToRequest(RespondWorkflowTaskCompletedRequest.Builder mutableRe
       ScheduleActivityTaskCommandAttributes commandAttributes =
           command.getScheduleActivityTaskCommandAttributes();
       if (!commandAttributes.getRequestEagerExecution()) continue;
-      boolean atLimit = this.reservedSlots.size() >= Config.EAGER_ACTIVITIES_LIMIT;
+      boolean atLimit = this.reservedSlots.size() >= this.maxReservations;
       Optional<SlotPermit> permit = Optional.empty();
       if (!atLimit) {
         permit = this.eagerActivityDispatcher.tryReserveActivitySlot(commandAttributes);

temporal-sdk/src/main/java/io/temporal/internal/worker/SyncWorkflowWorker.java

@@ -67,6 +67,7 @@ public SyncWorkflowWorker(
       String stickyTaskQueueName,
       @Nonnull WorkflowThreadExecutor workflowThreadExecutor,
       @Nonnull EagerActivityDispatcher eagerActivityDispatcher,
+      int maxEagerActivityReservationsPerWorkflowTask,
       @Nonnull SlotSupplier<WorkflowSlotInfo> slotSupplier,
       @Nonnull SlotSupplier<LocalActivitySlotInfo> laSlotSupplier,
       @Nonnull NamespaceCapabilities namespaceCapabilities) {
@@ -123,6 +124,7 @@ public SyncWorkflowWorker(
             cache,
             taskHandler,
             eagerActivityDispatcher,
+            maxEagerActivityReservationsPerWorkflowTask,
             slotSupplier,
             namespaceCapabilities);
 

temporal-sdk/src/main/java/io/temporal/internal/worker/WorkflowWorker.java

@@ -52,6 +52,7 @@ final class WorkflowWorker implements SuspendableWorker {
   private final Scope workerMetricsScope;
   private final GrpcRetryer grpcRetryer;
   private final EagerActivityDispatcher eagerActivityDispatcher;
+  private final int maxEagerActivityReservationsPerWorkflowTask;
   private final TrackingSlotSupplier<WorkflowSlotInfo> slotSupplier;
 
   private final TaskCounter taskCounter = new TaskCounter();
@@ -77,6 +78,7 @@ public WorkflowWorker(
       @Nonnull WorkflowExecutorCache cache,
       @Nonnull WorkflowTaskHandler handler,
       @Nonnull EagerActivityDispatcher eagerActivityDispatcher,
+      int maxEagerActivityReservationsPerWorkflowTask,
       @Nonnull SlotSupplier<WorkflowSlotInfo> slotSupplier,
       @Nonnull NamespaceCapabilities namespaceCapabilities) {
     this.service = Objects.requireNonNull(service);
@@ -92,6 +94,7 @@ public WorkflowWorker(
     this.handler = Objects.requireNonNull(handler);
     this.grpcRetryer = new GrpcRetryer(service.getServerCapabilities());
     this.eagerActivityDispatcher = eagerActivityDispatcher;
+    this.maxEagerActivityReservationsPerWorkflowTask = maxEagerActivityReservationsPerWorkflowTask;
     this.slotSupplier = new TrackingSlotSupplier<>(slotSupplier, this.workerMetricsScope);
     this.namespaceCapabilities = namespaceCapabilities;
   }
@@ -478,7 +481,8 @@ public void handle(WorkflowTask task) throws Exception {
                     RespondWorkflowTaskCompletedRequest.Builder requestBuilder =
                         taskCompleted.toBuilder();
                     try (EagerActivitySlotsReservation activitySlotsReservation =
-                        new EagerActivitySlotsReservation(eagerActivityDispatcher)) {
+                        new EagerActivitySlotsReservation(
+                            eagerActivityDispatcher, maxEagerActivityReservationsPerWorkflowTask)) {
                       activitySlotsReservation.applyToRequest(requestBuilder);
                       RespondWorkflowTaskCompletedResponse response =
                           sendTaskCompleted(

temporal-sdk/src/main/java/io/temporal/worker/Worker.java

@@ -244,6 +244,7 @@ private static final class TaskSnapshot {
             stickyTaskQueueName,
             workflowThreadExecutor,
             eagerActivityDispatcher,
+            this.options.getMaxEagerActivityReservationsPerWorkflowTask(),
             workflowSlotSupplier,
             localActivitySlotSupplier,
             namespaceCapabilities);

temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java

@@ -4,6 +4,7 @@
 
 import com.google.common.base.Preconditions;
 import io.temporal.common.Experimental;
+import io.temporal.internal.Config;
 import io.temporal.serviceclient.WorkflowServiceStubsOptions;
 import io.temporal.worker.tuning.*;
 import java.time.Duration;
@@ -64,6 +65,7 @@ public static final class Builder {
     private Duration defaultHeartbeatThrottleInterval;
     private Duration stickyQueueScheduleToStartTimeout;
     private boolean disableEagerExecution;
+    private int maxEagerActivityReservationsPerWorkflowTask = Config.EAGER_ACTIVITIES_LIMIT;
     private String buildId;
     private boolean useBuildIdForVersioning;
     private Duration stickyTaskQueueDrainTimeout;
@@ -109,6 +111,8 @@ private Builder(WorkerOptions o) {
       this.defaultHeartbeatThrottleInterval = o.defaultHeartbeatThrottleInterval;
       this.stickyQueueScheduleToStartTimeout = o.stickyQueueScheduleToStartTimeout;
       this.disableEagerExecution = o.disableEagerExecution;
+      this.maxEagerActivityReservationsPerWorkflowTask =
+          o.maxEagerActivityReservationsPerWorkflowTask;
       this.useBuildIdForVersioning = o.useBuildIdForVersioning;
       this.buildId = o.buildId;
       this.stickyTaskQueueDrainTimeout = o.stickyTaskQueueDrainTimeout;
@@ -400,6 +404,20 @@ public Builder setDisableEagerExecution(boolean disableEagerExecution) {
       return this;
     }
 
+    /**
+     * Sets the maximum number of activity slots that may be reserved for eager execution when
+     * completing a workflow task.
+     *
+     * <p>The default is 3. The value must be positive. To disable eager activity execution, use
+     * {@link #setDisableEagerExecution(boolean)}.
+     */
+    public Builder setMaxEagerActivityReservationsPerWorkflowTask(
+        int maxEagerActivityReservationsPerWorkflowTask) {
+      this.maxEagerActivityReservationsPerWorkflowTask =
+          maxEagerActivityReservationsPerWorkflowTask;
+      return this;
+    }
+
     /**
      * Opts the worker in to the Build-ID-based versioning feature. This ensures that the worker
      * will only receive tasks which it is compatible with.
@@ -623,6 +641,7 @@ public WorkerOptions build() {
           defaultHeartbeatThrottleInterval,
           stickyQueueScheduleToStartTimeout,
           disableEagerExecution,
+          maxEagerActivityReservationsPerWorkflowTask,
           useBuildIdForVersioning,
           buildId,
           stickyTaskQueueDrainTimeout,
@@ -647,6 +666,10 @@ public WorkerOptions validateAndBuildWithDefaults() {
           maxWorkerActivitiesPerSecond >= 0, "negative maxActivitiesPerSecond");
       Preconditions.checkState(
           maxConcurrentActivityExecutionSize >= 0, "negative maxConcurrentActivityExecutionSize");
+      Preconditions.checkState(
+          maxEagerActivityReservationsPerWorkflowTask > 0,
+          "maxEagerActivityReservationsPerWorkflowTask must be positive; use "
+              + "setDisableEagerExecution(true) to disable eager activity execution");
       Preconditions.checkState(
           maxConcurrentWorkflowTaskExecutionSize >= 0,
           "negative maxConcurrentWorkflowTaskExecutionSize");
@@ -758,6 +781,7 @@ public WorkerOptions validateAndBuildWithDefaults() {
               ? DEFAULT_STICKY_SCHEDULE_TO_START_TIMEOUT
               : stickyQueueScheduleToStartTimeout,
           disableEagerExecution,
+          maxEagerActivityReservationsPerWorkflowTask,
           useBuildIdForVersioning,
           buildId,
           stickyTaskQueueDrainTimeout == null
@@ -796,6 +820,7 @@ public WorkerOptions validateAndBuildWithDefaults() {
   private final Duration defaultHeartbeatThrottleInterval;
   private final @Nonnull Duration stickyQueueScheduleToStartTimeout;
   private final boolean disableEagerExecution;
+  private final int maxEagerActivityReservationsPerWorkflowTask;
   private final boolean useBuildIdForVersioning;
   private final String buildId;
   private final Duration stickyTaskQueueDrainTimeout;
@@ -831,6 +856,7 @@ private WorkerOptions(
       Duration defaultHeartbeatThrottleInterval,
       @Nonnull Duration stickyQueueScheduleToStartTimeout,
       boolean disableEagerExecution,
+      int maxEagerActivityReservationsPerWorkflowTask,
       boolean useBuildIdForVersioning,
       String buildId,
       Duration stickyTaskQueueDrainTimeout,
@@ -864,6 +890,7 @@ private WorkerOptions(
     this.defaultHeartbeatThrottleInterval = defaultHeartbeatThrottleInterval;
     this.stickyQueueScheduleToStartTimeout = stickyQueueScheduleToStartTimeout;
     this.disableEagerExecution = maxTaskQueueActivitiesPerSecond > 0 ? true : disableEagerExecution;
+    this.maxEagerActivityReservationsPerWorkflowTask = maxEagerActivityReservationsPerWorkflowTask;
     this.useBuildIdForVersioning = useBuildIdForVersioning;
     this.buildId = buildId;
     this.stickyTaskQueueDrainTimeout = stickyTaskQueueDrainTimeout;
@@ -989,6 +1016,10 @@ public boolean isEagerExecutionDisabled() {
     return disableEagerExecution;
   }
 
+  public int getMaxEagerActivityReservationsPerWorkflowTask() {
+    return maxEagerActivityReservationsPerWorkflowTask;
+  }
+
   public boolean isUsingBuildIdForVersioning() {
     return useBuildIdForVersioning;
   }
@@ -1070,6 +1101,8 @@ && compare(maxTaskQueueActivitiesPerSecond, that.maxTaskQueueActivitiesPerSecond
         && localActivityWorkerOnly == that.localActivityWorkerOnly
         && defaultDeadlockDetectionTimeout == that.defaultDeadlockDetectionTimeout
         && disableEagerExecution == that.disableEagerExecution
+        && maxEagerActivityReservationsPerWorkflowTask
+            == that.maxEagerActivityReservationsPerWorkflowTask
         && useBuildIdForVersioning == that.useBuildIdForVersioning
         && Objects.equals(workerTuner, that.workerTuner)
         && Objects.equals(maxHeartbeatThrottleInterval, that.maxHeartbeatThrottleInterval)
@@ -1109,6 +1142,7 @@ public int hashCode() {
         defaultHeartbeatThrottleInterval,
         stickyQueueScheduleToStartTimeout,
         disableEagerExecution,
+        maxEagerActivityReservationsPerWorkflowTask,
         useBuildIdForVersioning,
         buildId,
         stickyTaskQueueDrainTimeout,
@@ -1160,6 +1194,8 @@ public String toString() {
         + stickyQueueScheduleToStartTimeout
         + ", disableEagerExecution="
         + disableEagerExecution
+        + ", maxEagerActivityReservationsPerWorkflowTask="
+        + maxEagerActivityReservationsPerWorkflowTask
         + ", useBuildIdForVersioning="
         + useBuildIdForVersioning
         + ", buildId='"

temporal-sdk/src/test/java/io/temporal/internal/worker/EagerActivitySlotsReservationTest.java

@@ -0,0 +1,60 @@
+package io.temporal.internal.worker;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import io.temporal.api.command.v1.Command;
+import io.temporal.api.command.v1.ScheduleActivityTaskCommandAttributes;
+import io.temporal.api.enums.v1.CommandType;
+import io.temporal.api.workflowservice.v1.RespondWorkflowTaskCompletedRequest;
+import io.temporal.worker.tuning.SlotPermit;
+import java.util.Optional;
+import org.junit.Test;
+
+public class EagerActivitySlotsReservationTest {
+  @Test
+  public void limitsReservationsPerWorkflowTask() {
+    EagerActivityDispatcher dispatcher = mock(EagerActivityDispatcher.class);
+    when(dispatcher.tryReserveActivitySlot(any())).thenReturn(Optional.of(mock(SlotPermit.class)));
+    RespondWorkflowTaskCompletedRequest.Builder request =
+        RespondWorkflowTaskCompletedRequest.newBuilder();
+    for (int i = 0; i < 5; i++) {
+      request.addCommands(
+          Command.newBuilder()
+              .setCommandType(CommandType.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK)
+              .setScheduleActivityTaskCommandAttributes(
+                  ScheduleActivityTaskCommandAttributes.newBuilder()
+                      .setRequestEagerExecution(true)));
+    }
+
+    try (EagerActivitySlotsReservation reservation =
+        new EagerActivitySlotsReservation(dispatcher, 2)) {
+      reservation.applyToRequest(request);
+      assertEquals(5, request.getCommandsCount());
+      assertTrue(
+          request
+              .getCommands(0)
+              .getScheduleActivityTaskCommandAttributes()
+              .getRequestEagerExecution());
+      assertTrue(
+          request
+              .getCommands(1)
+              .getScheduleActivityTaskCommandAttributes()
+              .getRequestEagerExecution());
+      for (int i = 2; i < 5; i++) {
+        assertFalse(
+            request
+                .getCommands(i)
+                .getScheduleActivityTaskCommandAttributes()
+                .getRequestEagerExecution());
+      }
+    }
+    verify(dispatcher, times(2)).tryReserveActivitySlot(any());
+  }
+}

temporal-sdk/src/test/java/io/temporal/internal/worker/WorkflowWorkerTest.java

@@ -84,6 +84,7 @@ public void concurrentPollRequestLockTest() throws Exception {
             cache,
             taskHandler,
             eagerActivityDispatcher,
+            3,
             slotSupplier,
             new NamespaceCapabilities());
 
@@ -255,6 +256,7 @@ public void respondWorkflowTaskFailureMetricTest() throws Exception {
             cache,
             taskHandler,
             eagerActivityDispatcher,
+            3,
             slotSupplier,
             new NamespaceCapabilities());
 
@@ -399,6 +401,7 @@ public boolean isAnyTypeSupported() {
             cache,
             taskHandler,
             eagerActivityDispatcher,
+            3,
             slotSupplier,
             new NamespaceCapabilities());
 

temporal-sdk/src/test/java/io/temporal/worker/WorkerOptionsTest.java

@@ -23,6 +23,7 @@ public void build() {
   private void verifyBuild(WorkerOptions options) {
     assertEquals(10, options.getMaxConcurrentActivityExecutionSize());
     assertEquals(11, options.getMaxConcurrentLocalActivityExecutionSize());
+    assertEquals(3, options.getMaxEagerActivityReservationsPerWorkflowTask());
     assertNotNull(options.getPreferredVersionProvider());
   }
 
@@ -55,6 +56,7 @@ public void verifyNewBuilderFromExistingWorkerOptions() {
             .setDefaultHeartbeatThrottleInterval(Duration.ofSeconds(7))
             .setStickyQueueScheduleToStartTimeout(Duration.ofSeconds(60))
             .setDisableEagerExecution(false)
+            .setMaxEagerActivityReservationsPerWorkflowTask(17)
             .setUseBuildIdForVersioning(false)
             .setBuildId("build-id")
             .setStickyTaskQueueDrainTimeout(Duration.ofSeconds(15))
@@ -90,6 +92,9 @@ public void verifyNewBuilderFromExistingWorkerOptions() {
     assertEquals(
         w1.getStickyQueueScheduleToStartTimeout(), w2.getStickyQueueScheduleToStartTimeout());
     assertEquals(w1.isEagerExecutionDisabled(), w2.isEagerExecutionDisabled());
+    assertEquals(
+        w1.getMaxEagerActivityReservationsPerWorkflowTask(),
+        w2.getMaxEagerActivityReservationsPerWorkflowTask());
     assertEquals(w1.isUsingBuildIdForVersioning(), w2.isUsingBuildIdForVersioning());
     assertEquals(w1.getBuildId(), w2.getBuildId());
     assertEquals(w1.getStickyTaskQueueDrainTimeout(), w2.getStickyTaskQueueDrainTimeout());
@@ -230,4 +235,21 @@ public void verifyMaxTaskQueuePerSecondsDisablesEagerExecution() {
     WorkerOptions w2 = WorkerOptions.newBuilder().setMaxTaskQueueActivitiesPerSecond(2.0).build();
     assertTrue(w2.isEagerExecutionDisabled());
   }
+
+  @Test
+  public void rejectsNonPositiveMaxEagerActivityReservationsPerWorkflowTask() {
+    for (int value : new int[] {0, -1}) {
+      IllegalStateException exception =
+          assertThrows(
+              IllegalStateException.class,
+              () ->
+                  WorkerOptions.newBuilder()
+                      .setMaxEagerActivityReservationsPerWorkflowTask(value)
+                      .validateAndBuildWithDefaults());
+      assertEquals(
+          "maxEagerActivityReservationsPerWorkflowTask must be positive; use "
+              + "setDisableEagerExecution(true) to disable eager activity execution",
+          exception.getMessage());
+    }
+  }
 }