diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/EagerActivitySlotsReservation.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/EagerActivitySlotsReservation.java index 29c50bb47..9f84db488 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/EagerActivitySlotsReservation.java +++ b/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 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 permit = Optional.empty(); if (!atLimit) { permit = this.eagerActivityDispatcher.tryReserveActivitySlot(commandAttributes); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncWorkflowWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncWorkflowWorker.java index 18cf7fd4a..be128a5e6 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncWorkflowWorker.java +++ b/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 slotSupplier, @Nonnull SlotSupplier laSlotSupplier, @Nonnull NamespaceCapabilities namespaceCapabilities) { @@ -123,6 +124,7 @@ public SyncWorkflowWorker( cache, taskHandler, eagerActivityDispatcher, + maxEagerActivityReservationsPerWorkflowTask, slotSupplier, namespaceCapabilities); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkflowWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkflowWorker.java index b86dfea6d..3eed1099d 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/WorkflowWorker.java +++ b/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 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 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( diff --git a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java index 93a26def2..b75513444 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java +++ b/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); diff --git a/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java b/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java index aa61c6fa4..3346b5314 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java +++ b/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. + * + *

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='" diff --git a/temporal-sdk/src/test/java/io/temporal/internal/worker/EagerActivitySlotsReservationTest.java b/temporal-sdk/src/test/java/io/temporal/internal/worker/EagerActivitySlotsReservationTest.java new file mode 100644 index 000000000..7dff989c7 --- /dev/null +++ b/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()); + } +} diff --git a/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkflowWorkerTest.java b/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkflowWorkerTest.java index d4f1824c2..5cd1fc8d3 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/worker/WorkflowWorkerTest.java +++ b/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()); diff --git a/temporal-sdk/src/test/java/io/temporal/worker/WorkerOptionsTest.java b/temporal-sdk/src/test/java/io/temporal/worker/WorkerOptionsTest.java index f21345a30..1dd61df1d 100644 --- a/temporal-sdk/src/test/java/io/temporal/worker/WorkerOptionsTest.java +++ b/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()); + } + } } diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/activityTests/EagerActivityDispatchingTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/activityTests/EagerActivityDispatchingTest.java index 3baf855fb..8353b40c3 100644 --- a/temporal-sdk/src/test/java/io/temporal/workflow/activityTests/EagerActivityDispatchingTest.java +++ b/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,8 +32,10 @@ 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.*; @@ -32,6 +43,8 @@ public class EagerActivityDispatchingTest { private static final String TASK_QUEUE = "test-eager-activity-dispatch"; private TestWorkflowEnvironment env; private ArrayList workerFactories; + private final EagerActivityRequestInterceptor eagerActivityRequestInterceptor = + new EagerActivityRequestInterceptor(); private final TestActivitiesImpl activitiesImpl = new TestActivitiesImpl(); CountingSlotSupplier 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 ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + if (method == WorkflowServiceGrpc.getRespondWorkflowTaskCompletedMethod()) { + return new ForwardingClientCall.SimpleForwardingClientCall( + 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); + } + } }