Validate eager activity reservation limits

Author: Sushisource

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

@@ -408,7 +408,8 @@ public Builder setDisableEagerExecution(boolean disableEagerExecution) {
      * Sets the maximum number of activity slots that may be reserved for eager execution when
      * completing a workflow task.
      *
-     * <p>The default is 3. Setting this to zero disables eager activity execution.
+     * <p>The default is 3. The value must be positive. To disable eager activity execution, use
+     * {@link #setDisableEagerExecution(boolean)}.
      */
     public Builder setMaxEagerActivityReservationsPerWorkflowTask(
         int maxEagerActivityReservationsPerWorkflowTask) {
@@ -666,8 +667,9 @@ public WorkerOptions validateAndBuildWithDefaults() {
       Preconditions.checkState(
           maxConcurrentActivityExecutionSize >= 0, "negative maxConcurrentActivityExecutionSize");
       Preconditions.checkState(
-          maxEagerActivityReservationsPerWorkflowTask >= 0,
-          "negative maxEagerActivityReservationsPerWorkflowTask");
+          maxEagerActivityReservationsPerWorkflowTask > 0,
+          "maxEagerActivityReservationsPerWorkflowTask must be positive; use "
+              + "setDisableEagerExecution(true) to disable eager activity execution");
       Preconditions.checkState(
           maxConcurrentWorkflowTaskExecutionSize >= 0,
           "negative maxConcurrentWorkflowTaskExecutionSize");

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

@@ -235,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());
+    }
+  }
 }

temporal-sdk/src/test/java/io/temporal/workflow/activityTests/EagerActivityDispatchingTest.java

@@ -3,13 +3,22 @@
 import static org.junit.Assert.*;
 import static org.junit.Assume.*;
 
+import io.grpc.CallOptions;
+import io.grpc.Channel;
+import io.grpc.ClientCall;
+import io.grpc.ClientInterceptor;
+import io.grpc.ForwardingClientCall;
+import io.grpc.MethodDescriptor;
 import io.temporal.activity.ActivityOptions;
 import io.temporal.api.history.v1.HistoryEvent;
+import io.temporal.api.workflowservice.v1.RespondWorkflowTaskCompletedRequest;
+import io.temporal.api.workflowservice.v1.WorkflowServiceGrpc;
 import io.temporal.client.WorkflowClient;
 import io.temporal.client.WorkflowOptions;
 import io.temporal.client.WorkflowStub;
 import io.temporal.common.WorkflowExecutionHistory;
 import io.temporal.internal.Config;
+import io.temporal.serviceclient.WorkflowServiceStubsOptions;
 import io.temporal.testUtils.CountingSlotSupplier;
 import io.temporal.testing.TestWorkflowEnvironment;
 import io.temporal.testing.internal.ExternalServiceTestConfigurator;
@@ -23,15 +32,19 @@
 import io.temporal.workflow.shared.TestActivities.TestActivitiesImpl;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import org.junit.*;
 
 public class EagerActivityDispatchingTest {
   private static final String TASK_QUEUE = "test-eager-activity-dispatch";
   private TestWorkflowEnvironment env;
   private ArrayList<WorkerFactory> workerFactories;
+  private final EagerActivityRequestInterceptor eagerActivityRequestInterceptor =
+      new EagerActivityRequestInterceptor();
 
   private final TestActivitiesImpl activitiesImpl = new TestActivitiesImpl();
   CountingSlotSupplier<WorkflowSlotInfo> workflowTaskSlotSupplier = new CountingSlotSupplier<>(100);
@@ -42,9 +55,16 @@ public class EagerActivityDispatchingTest {
 
   @Before
   public void setUp() throws Exception {
+    eagerActivityRequestInterceptor.reset();
     this.env =
         TestWorkflowEnvironment.newInstance(
-            ExternalServiceTestConfigurator.configuredTestEnvironmentOptions().build());
+            ExternalServiceTestConfigurator.configuredTestEnvironmentOptions()
+                .setWorkflowServiceStubsOptions(
+                    WorkflowServiceStubsOptions.newBuilder()
+                        .setGrpcClientInterceptors(
+                            Collections.singletonList(eagerActivityRequestInterceptor))
+                        .build())
+                .build());
     this.workerFactories = new ArrayList<>();
   }
 
@@ -125,6 +145,25 @@ public void testEagerActivities() {
     assertFalse(activityTaskStartedEventIdentity.contains("worker2"));
   }
 
+  @Test
+  public void testMaxEagerActivityReservationsPerWorkflowTask() {
+    setupWorker(
+        "worker1",
+        WorkerOptions.newBuilder()
+            .setMaxEagerActivityReservationsPerWorkflowTask(2)
+            .setDisableEagerExecution(false),
+        true);
+
+    EagerActivityTestWorkflow workflowStub =
+        env.getWorkflowClient()
+            .newWorkflowStub(
+                EagerActivityTestWorkflow.class,
+                WorkflowOptions.newBuilder().setTaskQueue(TASK_QUEUE).build());
+    workflowStub.execute(true);
+
+    assertEquals(2, eagerActivityRequestInterceptor.getEagerActivityRequestCount());
+  }
+
   @Test
   public void testNoEagerActivitiesIfDisabledOnWorker() {
     assumeTrue(
@@ -222,4 +261,50 @@ public void execute(boolean enableEagerActivityDispatch) {
       Promise.allOf(promises).get();
     }
   }
+
+  private static class EagerActivityRequestInterceptor implements ClientInterceptor {
+    private final AtomicInteger eagerActivityRequestCount = new AtomicInteger(-1);
+
+    @Override
+    public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+        MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
+      if (method == WorkflowServiceGrpc.getRespondWorkflowTaskCompletedMethod()) {
+        return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(
+            next.newCall(method, callOptions)) {
+          @Override
+          public void sendMessage(ReqT message) {
+            RespondWorkflowTaskCompletedRequest request =
+                (RespondWorkflowTaskCompletedRequest) message;
+            long activityCommandCount =
+                request.getCommandsList().stream()
+                    .filter(command -> command.hasScheduleActivityTaskCommandAttributes())
+                    .count();
+            if (activityCommandCount > 0) {
+              int eagerRequestCount =
+                  (int)
+                      request.getCommandsList().stream()
+                          .filter(command -> command.hasScheduleActivityTaskCommandAttributes())
+                          .filter(
+                              command ->
+                                  command
+                                      .getScheduleActivityTaskCommandAttributes()
+                                      .getRequestEagerExecution())
+                          .count();
+              eagerActivityRequestCount.compareAndSet(-1, eagerRequestCount);
+            }
+            super.sendMessage(message);
+          }
+        };
+      }
+      return next.newCall(method, callOptions);
+    }
+
+    int getEagerActivityRequestCount() {
+      return eagerActivityRequestCount.get();
+    }
+
+    void reset() {
+      eagerActivityRequestCount.set(-1);
+    }
+  }
 }