From 31caa47a7e8ce9862a4fc5039cf81b381baaf847 Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Fri, 24 Jul 2026 11:32:11 -0700 Subject: [PATCH 1/3] Expose eager activity concurrency limit --- .../internal/worker/ActivityWorker.java | 70 ++++++++++++++++--- .../internal/worker/SyncActivityWorker.java | 5 +- .../main/java/io/temporal/worker/Worker.java | 3 +- .../io/temporal/worker/WorkerOptions.java | 31 ++++++++ .../worker/EagerActivitySlotLimiterTest.java | 31 ++++++++ .../io/temporal/worker/WorkerOptionsTest.java | 4 ++ 6 files changed, 133 insertions(+), 11 deletions(-) create mode 100644 temporal-sdk/src/test/java/io/temporal/internal/worker/EagerActivitySlotLimiterTest.java diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java index ff528d46b..f0c1c9a5c 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java @@ -218,8 +218,9 @@ public WorkerLifecycleState getLifecycleState() { return poller.getLifecycleState(); } - public EagerActivityDispatcher getEagerActivityDispatcher() { - return new EagerActivityDispatcherImpl(); + public EagerActivityDispatcher getEagerActivityDispatcher( + int maxConcurrentEagerActivityExecutionSize) { + return new EagerActivityDispatcherImpl(maxConcurrentEagerActivityExecutionSize); } private PollerOptions getPollerOptions(SingleWorkerOptions options) { @@ -485,6 +486,13 @@ private void logExceptionDuringResultReporting( } private final class EagerActivityDispatcherImpl implements EagerActivityDispatcher { + private final EagerActivitySlotLimiter eagerActivitySlotLimiter; + + private EagerActivityDispatcherImpl(int maxConcurrentEagerActivityExecutionSize) { + this.eagerActivitySlotLimiter = + new EagerActivitySlotLimiter(maxConcurrentEagerActivityExecutionSize); + } + @Override public Optional tryReserveActivitySlot( ScheduleActivityTaskCommandAttributesOrBuilder commandAttributes) { @@ -493,15 +501,31 @@ public Optional tryReserveActivitySlot( commandAttributes.getTaskQueue().getName(), ActivityWorker.this.taskQueue)) { return Optional.empty(); } - return ActivityWorker.this.slotSupplier.tryReserveSlot( - new SlotReservationData( - ActivityWorker.this.taskQueue, options.getIdentity(), options.getBuildId())); + if (!eagerActivitySlotLimiter.tryReserve()) { + return Optional.empty(); + } + Optional permit = Optional.empty(); + try { + permit = + ActivityWorker.this.slotSupplier.tryReserveSlot( + new SlotReservationData( + ActivityWorker.this.taskQueue, options.getIdentity(), options.getBuildId())); + return permit; + } finally { + if (!permit.isPresent()) { + eagerActivitySlotLimiter.release(); + } + } } @Override public void releaseActivitySlotReservations(Iterable permits) { for (SlotPermit permit : permits) { - ActivityWorker.this.slotSupplier.releaseSlot(SlotReleaseReason.neverUsed(), permit); + try { + ActivityWorker.this.slotSupplier.releaseSlot(SlotReleaseReason.neverUsed(), permit); + } finally { + eagerActivitySlotLimiter.release(); + } } } @@ -511,9 +535,39 @@ public void dispatchActivity(PollActivityTaskQueueResponse activity, SlotPermit new ActivityTask( activity, permit, - () -> + () -> { + try { ActivityWorker.this.slotSupplier.releaseSlot( - SlotReleaseReason.taskComplete(), permit))); + SlotReleaseReason.taskComplete(), permit); + } finally { + eagerActivitySlotLimiter.release(); + } + })); + } + } + + static final class EagerActivitySlotLimiter { + private final int maxConcurrent; + private int heldSlotCount; + + EagerActivitySlotLimiter(int maxConcurrent) { + this.maxConcurrent = maxConcurrent; + } + + synchronized boolean tryReserve() { + if (maxConcurrent > 0 && heldSlotCount >= maxConcurrent) { + return false; + } + heldSlotCount++; + return true; + } + + synchronized void release() { + if (heldSlotCount <= 0) { + throw new IllegalStateException( + "Trying to release an unreserved eager activity slot. This is an SDK bug."); + } + heldSlotCount--; } } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncActivityWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncActivityWorker.java index 94d2f5dee..e1c75f765 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncActivityWorker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncActivityWorker.java @@ -163,8 +163,9 @@ public WorkerLifecycleState getLifecycleState() { } } - public EagerActivityDispatcher getEagerActivityDispatcher() { - return this.worker.getEagerActivityDispatcher(); + public EagerActivityDispatcher getEagerActivityDispatcher( + int maxConcurrentEagerActivityExecutionSize) { + return this.worker.getEagerActivityDispatcher(maxConcurrentEagerActivityExecutionSize); } public boolean isAnyTypeSupported() { 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..8420e8eb0 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java @@ -173,7 +173,8 @@ private static final class TaskSnapshot { EagerActivityDispatcher eagerActivityDispatcher = (activityWorker != null && !this.options.isEagerExecutionDisabled()) - ? activityWorker.getEagerActivityDispatcher() + ? activityWorker.getEagerActivityDispatcher( + this.options.getMaxConcurrentEagerActivityExecutionSize()) : new EagerActivityDispatcher.NoopEagerActivityDispatcher(); SingleWorkerOptions nexusOptions = 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..b0614dafe 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java @@ -64,6 +64,7 @@ public static final class Builder { private Duration defaultHeartbeatThrottleInterval; private Duration stickyQueueScheduleToStartTimeout; private boolean disableEagerExecution; + private int maxConcurrentEagerActivityExecutionSize; private String buildId; private boolean useBuildIdForVersioning; private Duration stickyTaskQueueDrainTimeout; @@ -109,6 +110,7 @@ private Builder(WorkerOptions o) { this.defaultHeartbeatThrottleInterval = o.defaultHeartbeatThrottleInterval; this.stickyQueueScheduleToStartTimeout = o.stickyQueueScheduleToStartTimeout; this.disableEagerExecution = o.disableEagerExecution; + this.maxConcurrentEagerActivityExecutionSize = o.maxConcurrentEagerActivityExecutionSize; this.useBuildIdForVersioning = o.useBuildIdForVersioning; this.buildId = o.buildId; this.stickyTaskQueueDrainTimeout = o.stickyTaskQueueDrainTimeout; @@ -400,6 +402,19 @@ public Builder setDisableEagerExecution(boolean disableEagerExecution) { return this; } + /** + * Sets the maximum number of eager activities that can be running concurrently. + * + *

When nonzero, eager activity execution will not be requested if it would cause the number + * of running eager activities to exceed this value. The default of zero means unlimited and + * therefore only bound by the activity slot supplier. + */ + public Builder setMaxConcurrentEagerActivityExecutionSize( + int maxConcurrentEagerActivityExecutionSize) { + this.maxConcurrentEagerActivityExecutionSize = maxConcurrentEagerActivityExecutionSize; + 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 +638,7 @@ public WorkerOptions build() { defaultHeartbeatThrottleInterval, stickyQueueScheduleToStartTimeout, disableEagerExecution, + maxConcurrentEagerActivityExecutionSize, useBuildIdForVersioning, buildId, stickyTaskQueueDrainTimeout, @@ -647,6 +663,9 @@ public WorkerOptions validateAndBuildWithDefaults() { maxWorkerActivitiesPerSecond >= 0, "negative maxActivitiesPerSecond"); Preconditions.checkState( maxConcurrentActivityExecutionSize >= 0, "negative maxConcurrentActivityExecutionSize"); + Preconditions.checkState( + maxConcurrentEagerActivityExecutionSize >= 0, + "negative maxConcurrentEagerActivityExecutionSize"); Preconditions.checkState( maxConcurrentWorkflowTaskExecutionSize >= 0, "negative maxConcurrentWorkflowTaskExecutionSize"); @@ -758,6 +777,7 @@ public WorkerOptions validateAndBuildWithDefaults() { ? DEFAULT_STICKY_SCHEDULE_TO_START_TIMEOUT : stickyQueueScheduleToStartTimeout, disableEagerExecution, + maxConcurrentEagerActivityExecutionSize, useBuildIdForVersioning, buildId, stickyTaskQueueDrainTimeout == null @@ -796,6 +816,7 @@ public WorkerOptions validateAndBuildWithDefaults() { private final Duration defaultHeartbeatThrottleInterval; private final @Nonnull Duration stickyQueueScheduleToStartTimeout; private final boolean disableEagerExecution; + private final int maxConcurrentEagerActivityExecutionSize; private final boolean useBuildIdForVersioning; private final String buildId; private final Duration stickyTaskQueueDrainTimeout; @@ -831,6 +852,7 @@ private WorkerOptions( Duration defaultHeartbeatThrottleInterval, @Nonnull Duration stickyQueueScheduleToStartTimeout, boolean disableEagerExecution, + int maxConcurrentEagerActivityExecutionSize, boolean useBuildIdForVersioning, String buildId, Duration stickyTaskQueueDrainTimeout, @@ -864,6 +886,7 @@ private WorkerOptions( this.defaultHeartbeatThrottleInterval = defaultHeartbeatThrottleInterval; this.stickyQueueScheduleToStartTimeout = stickyQueueScheduleToStartTimeout; this.disableEagerExecution = maxTaskQueueActivitiesPerSecond > 0 ? true : disableEagerExecution; + this.maxConcurrentEagerActivityExecutionSize = maxConcurrentEagerActivityExecutionSize; this.useBuildIdForVersioning = useBuildIdForVersioning; this.buildId = buildId; this.stickyTaskQueueDrainTimeout = stickyTaskQueueDrainTimeout; @@ -989,6 +1012,10 @@ public boolean isEagerExecutionDisabled() { return disableEagerExecution; } + public int getMaxConcurrentEagerActivityExecutionSize() { + return maxConcurrentEagerActivityExecutionSize; + } + public boolean isUsingBuildIdForVersioning() { return useBuildIdForVersioning; } @@ -1070,6 +1097,7 @@ && compare(maxTaskQueueActivitiesPerSecond, that.maxTaskQueueActivitiesPerSecond && localActivityWorkerOnly == that.localActivityWorkerOnly && defaultDeadlockDetectionTimeout == that.defaultDeadlockDetectionTimeout && disableEagerExecution == that.disableEagerExecution + && maxConcurrentEagerActivityExecutionSize == that.maxConcurrentEagerActivityExecutionSize && useBuildIdForVersioning == that.useBuildIdForVersioning && Objects.equals(workerTuner, that.workerTuner) && Objects.equals(maxHeartbeatThrottleInterval, that.maxHeartbeatThrottleInterval) @@ -1109,6 +1137,7 @@ public int hashCode() { defaultHeartbeatThrottleInterval, stickyQueueScheduleToStartTimeout, disableEagerExecution, + maxConcurrentEagerActivityExecutionSize, useBuildIdForVersioning, buildId, stickyTaskQueueDrainTimeout, @@ -1160,6 +1189,8 @@ public String toString() { + stickyQueueScheduleToStartTimeout + ", disableEagerExecution=" + disableEagerExecution + + ", maxConcurrentEagerActivityExecutionSize=" + + maxConcurrentEagerActivityExecutionSize + ", useBuildIdForVersioning=" + useBuildIdForVersioning + ", buildId='" diff --git a/temporal-sdk/src/test/java/io/temporal/internal/worker/EagerActivitySlotLimiterTest.java b/temporal-sdk/src/test/java/io/temporal/internal/worker/EagerActivitySlotLimiterTest.java new file mode 100644 index 000000000..41d5a5dea --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/worker/EagerActivitySlotLimiterTest.java @@ -0,0 +1,31 @@ +package io.temporal.internal.worker; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import org.junit.Test; + +public class EagerActivitySlotLimiterTest { + @Test + public void enforcesLimitUntilReservationIsReleased() { + ActivityWorker.EagerActivitySlotLimiter limiter = + new ActivityWorker.EagerActivitySlotLimiter(2); + + assertTrue(limiter.tryReserve()); + assertTrue(limiter.tryReserve()); + assertFalse(limiter.tryReserve()); + + limiter.release(); + assertTrue(limiter.tryReserve()); + } + + @Test + public void zeroAllowsUnlimitedReservations() { + ActivityWorker.EagerActivitySlotLimiter limiter = + new ActivityWorker.EagerActivitySlotLimiter(0); + + for (int i = 0; i < 1000; i++) { + assertTrue(limiter.tryReserve()); + } + } +} 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..62894f10c 100644 --- a/temporal-sdk/src/test/java/io/temporal/worker/WorkerOptionsTest.java +++ b/temporal-sdk/src/test/java/io/temporal/worker/WorkerOptionsTest.java @@ -55,6 +55,7 @@ public void verifyNewBuilderFromExistingWorkerOptions() { .setDefaultHeartbeatThrottleInterval(Duration.ofSeconds(7)) .setStickyQueueScheduleToStartTimeout(Duration.ofSeconds(60)) .setDisableEagerExecution(false) + .setMaxConcurrentEagerActivityExecutionSize(17) .setUseBuildIdForVersioning(false) .setBuildId("build-id") .setStickyTaskQueueDrainTimeout(Duration.ofSeconds(15)) @@ -90,6 +91,9 @@ public void verifyNewBuilderFromExistingWorkerOptions() { assertEquals( w1.getStickyQueueScheduleToStartTimeout(), w2.getStickyQueueScheduleToStartTimeout()); assertEquals(w1.isEagerExecutionDisabled(), w2.isEagerExecutionDisabled()); + assertEquals( + w1.getMaxConcurrentEagerActivityExecutionSize(), + w2.getMaxConcurrentEagerActivityExecutionSize()); assertEquals(w1.isUsingBuildIdForVersioning(), w2.isUsingBuildIdForVersioning()); assertEquals(w1.getBuildId(), w2.getBuildId()); assertEquals(w1.getStickyTaskQueueDrainTimeout(), w2.getStickyTaskQueueDrainTimeout()); From e4a9841150b4c7cbc956d959820af62fc9522c08 Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Fri, 24 Jul 2026 12:14:07 -0700 Subject: [PATCH 2/3] Configure eager reservations per workflow task --- .../internal/worker/ActivityWorker.java | 70 +++---------------- .../worker/EagerActivitySlotsReservation.java | 8 ++- .../internal/worker/SyncActivityWorker.java | 5 +- .../internal/worker/SyncWorkflowWorker.java | 2 + .../internal/worker/WorkflowWorker.java | 6 +- .../main/java/io/temporal/worker/Worker.java | 4 +- .../io/temporal/worker/WorkerOptions.java | 47 +++++++------ .../worker/EagerActivitySlotLimiterTest.java | 31 -------- .../EagerActivitySlotsReservationTest.java | 60 ++++++++++++++++ .../internal/worker/WorkflowWorkerTest.java | 3 + .../io/temporal/worker/WorkerOptionsTest.java | 7 +- 11 files changed, 116 insertions(+), 127 deletions(-) delete mode 100644 temporal-sdk/src/test/java/io/temporal/internal/worker/EagerActivitySlotLimiterTest.java create mode 100644 temporal-sdk/src/test/java/io/temporal/internal/worker/EagerActivitySlotsReservationTest.java diff --git a/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java index f0c1c9a5c..ff528d46b 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java @@ -218,9 +218,8 @@ public WorkerLifecycleState getLifecycleState() { return poller.getLifecycleState(); } - public EagerActivityDispatcher getEagerActivityDispatcher( - int maxConcurrentEagerActivityExecutionSize) { - return new EagerActivityDispatcherImpl(maxConcurrentEagerActivityExecutionSize); + public EagerActivityDispatcher getEagerActivityDispatcher() { + return new EagerActivityDispatcherImpl(); } private PollerOptions getPollerOptions(SingleWorkerOptions options) { @@ -486,13 +485,6 @@ private void logExceptionDuringResultReporting( } private final class EagerActivityDispatcherImpl implements EagerActivityDispatcher { - private final EagerActivitySlotLimiter eagerActivitySlotLimiter; - - private EagerActivityDispatcherImpl(int maxConcurrentEagerActivityExecutionSize) { - this.eagerActivitySlotLimiter = - new EagerActivitySlotLimiter(maxConcurrentEagerActivityExecutionSize); - } - @Override public Optional tryReserveActivitySlot( ScheduleActivityTaskCommandAttributesOrBuilder commandAttributes) { @@ -501,31 +493,15 @@ public Optional tryReserveActivitySlot( commandAttributes.getTaskQueue().getName(), ActivityWorker.this.taskQueue)) { return Optional.empty(); } - if (!eagerActivitySlotLimiter.tryReserve()) { - return Optional.empty(); - } - Optional permit = Optional.empty(); - try { - permit = - ActivityWorker.this.slotSupplier.tryReserveSlot( - new SlotReservationData( - ActivityWorker.this.taskQueue, options.getIdentity(), options.getBuildId())); - return permit; - } finally { - if (!permit.isPresent()) { - eagerActivitySlotLimiter.release(); - } - } + return ActivityWorker.this.slotSupplier.tryReserveSlot( + new SlotReservationData( + ActivityWorker.this.taskQueue, options.getIdentity(), options.getBuildId())); } @Override public void releaseActivitySlotReservations(Iterable permits) { for (SlotPermit permit : permits) { - try { - ActivityWorker.this.slotSupplier.releaseSlot(SlotReleaseReason.neverUsed(), permit); - } finally { - eagerActivitySlotLimiter.release(); - } + ActivityWorker.this.slotSupplier.releaseSlot(SlotReleaseReason.neverUsed(), permit); } } @@ -535,39 +511,9 @@ public void dispatchActivity(PollActivityTaskQueueResponse activity, SlotPermit new ActivityTask( activity, permit, - () -> { - try { + () -> ActivityWorker.this.slotSupplier.releaseSlot( - SlotReleaseReason.taskComplete(), permit); - } finally { - eagerActivitySlotLimiter.release(); - } - })); - } - } - - static final class EagerActivitySlotLimiter { - private final int maxConcurrent; - private int heldSlotCount; - - EagerActivitySlotLimiter(int maxConcurrent) { - this.maxConcurrent = maxConcurrent; - } - - synchronized boolean tryReserve() { - if (maxConcurrent > 0 && heldSlotCount >= maxConcurrent) { - return false; - } - heldSlotCount++; - return true; - } - - synchronized void release() { - if (heldSlotCount <= 0) { - throw new IllegalStateException( - "Trying to release an unreserved eager activity slot. This is an SDK bug."); - } - heldSlotCount--; + SlotReleaseReason.taskComplete(), permit))); } } } 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/SyncActivityWorker.java b/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncActivityWorker.java index e1c75f765..94d2f5dee 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncActivityWorker.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/worker/SyncActivityWorker.java @@ -163,9 +163,8 @@ public WorkerLifecycleState getLifecycleState() { } } - public EagerActivityDispatcher getEagerActivityDispatcher( - int maxConcurrentEagerActivityExecutionSize) { - return this.worker.getEagerActivityDispatcher(maxConcurrentEagerActivityExecutionSize); + public EagerActivityDispatcher getEagerActivityDispatcher() { + return this.worker.getEagerActivityDispatcher(); } public boolean isAnyTypeSupported() { 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 8420e8eb0..b75513444 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/Worker.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/Worker.java @@ -173,8 +173,7 @@ private static final class TaskSnapshot { EagerActivityDispatcher eagerActivityDispatcher = (activityWorker != null && !this.options.isEagerExecutionDisabled()) - ? activityWorker.getEagerActivityDispatcher( - this.options.getMaxConcurrentEagerActivityExecutionSize()) + ? activityWorker.getEagerActivityDispatcher() : new EagerActivityDispatcher.NoopEagerActivityDispatcher(); SingleWorkerOptions nexusOptions = @@ -245,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 b0614dafe..73bb0ca8f 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,7 +65,7 @@ public static final class Builder { private Duration defaultHeartbeatThrottleInterval; private Duration stickyQueueScheduleToStartTimeout; private boolean disableEagerExecution; - private int maxConcurrentEagerActivityExecutionSize; + private int maxEagerActivityReservationsPerWorkflowTask = Config.EAGER_ACTIVITIES_LIMIT; private String buildId; private boolean useBuildIdForVersioning; private Duration stickyTaskQueueDrainTimeout; @@ -110,7 +111,8 @@ private Builder(WorkerOptions o) { this.defaultHeartbeatThrottleInterval = o.defaultHeartbeatThrottleInterval; this.stickyQueueScheduleToStartTimeout = o.stickyQueueScheduleToStartTimeout; this.disableEagerExecution = o.disableEagerExecution; - this.maxConcurrentEagerActivityExecutionSize = o.maxConcurrentEagerActivityExecutionSize; + this.maxEagerActivityReservationsPerWorkflowTask = + o.maxEagerActivityReservationsPerWorkflowTask; this.useBuildIdForVersioning = o.useBuildIdForVersioning; this.buildId = o.buildId; this.stickyTaskQueueDrainTimeout = o.stickyTaskQueueDrainTimeout; @@ -403,15 +405,15 @@ public Builder setDisableEagerExecution(boolean disableEagerExecution) { } /** - * Sets the maximum number of eager activities that can be running concurrently. + * Sets the maximum number of activity slots that may be reserved for eager execution when + * completing a workflow task. * - *

When nonzero, eager activity execution will not be requested if it would cause the number - * of running eager activities to exceed this value. The default of zero means unlimited and - * therefore only bound by the activity slot supplier. + *

The default is 3. Setting this to zero disables eager activity execution. */ - public Builder setMaxConcurrentEagerActivityExecutionSize( - int maxConcurrentEagerActivityExecutionSize) { - this.maxConcurrentEagerActivityExecutionSize = maxConcurrentEagerActivityExecutionSize; + public Builder setMaxEagerActivityReservationsPerWorkflowTask( + int maxEagerActivityReservationsPerWorkflowTask) { + this.maxEagerActivityReservationsPerWorkflowTask = + maxEagerActivityReservationsPerWorkflowTask; return this; } @@ -638,7 +640,7 @@ public WorkerOptions build() { defaultHeartbeatThrottleInterval, stickyQueueScheduleToStartTimeout, disableEagerExecution, - maxConcurrentEagerActivityExecutionSize, + maxEagerActivityReservationsPerWorkflowTask, useBuildIdForVersioning, buildId, stickyTaskQueueDrainTimeout, @@ -664,8 +666,8 @@ public WorkerOptions validateAndBuildWithDefaults() { Preconditions.checkState( maxConcurrentActivityExecutionSize >= 0, "negative maxConcurrentActivityExecutionSize"); Preconditions.checkState( - maxConcurrentEagerActivityExecutionSize >= 0, - "negative maxConcurrentEagerActivityExecutionSize"); + maxEagerActivityReservationsPerWorkflowTask >= 0, + "negative maxEagerActivityReservationsPerWorkflowTask"); Preconditions.checkState( maxConcurrentWorkflowTaskExecutionSize >= 0, "negative maxConcurrentWorkflowTaskExecutionSize"); @@ -777,7 +779,7 @@ public WorkerOptions validateAndBuildWithDefaults() { ? DEFAULT_STICKY_SCHEDULE_TO_START_TIMEOUT : stickyQueueScheduleToStartTimeout, disableEagerExecution, - maxConcurrentEagerActivityExecutionSize, + maxEagerActivityReservationsPerWorkflowTask, useBuildIdForVersioning, buildId, stickyTaskQueueDrainTimeout == null @@ -816,7 +818,7 @@ public WorkerOptions validateAndBuildWithDefaults() { private final Duration defaultHeartbeatThrottleInterval; private final @Nonnull Duration stickyQueueScheduleToStartTimeout; private final boolean disableEagerExecution; - private final int maxConcurrentEagerActivityExecutionSize; + private final int maxEagerActivityReservationsPerWorkflowTask; private final boolean useBuildIdForVersioning; private final String buildId; private final Duration stickyTaskQueueDrainTimeout; @@ -852,7 +854,7 @@ private WorkerOptions( Duration defaultHeartbeatThrottleInterval, @Nonnull Duration stickyQueueScheduleToStartTimeout, boolean disableEagerExecution, - int maxConcurrentEagerActivityExecutionSize, + int maxEagerActivityReservationsPerWorkflowTask, boolean useBuildIdForVersioning, String buildId, Duration stickyTaskQueueDrainTimeout, @@ -886,7 +888,7 @@ private WorkerOptions( this.defaultHeartbeatThrottleInterval = defaultHeartbeatThrottleInterval; this.stickyQueueScheduleToStartTimeout = stickyQueueScheduleToStartTimeout; this.disableEagerExecution = maxTaskQueueActivitiesPerSecond > 0 ? true : disableEagerExecution; - this.maxConcurrentEagerActivityExecutionSize = maxConcurrentEagerActivityExecutionSize; + this.maxEagerActivityReservationsPerWorkflowTask = maxEagerActivityReservationsPerWorkflowTask; this.useBuildIdForVersioning = useBuildIdForVersioning; this.buildId = buildId; this.stickyTaskQueueDrainTimeout = stickyTaskQueueDrainTimeout; @@ -1012,8 +1014,8 @@ public boolean isEagerExecutionDisabled() { return disableEagerExecution; } - public int getMaxConcurrentEagerActivityExecutionSize() { - return maxConcurrentEagerActivityExecutionSize; + public int getMaxEagerActivityReservationsPerWorkflowTask() { + return maxEagerActivityReservationsPerWorkflowTask; } public boolean isUsingBuildIdForVersioning() { @@ -1097,7 +1099,8 @@ && compare(maxTaskQueueActivitiesPerSecond, that.maxTaskQueueActivitiesPerSecond && localActivityWorkerOnly == that.localActivityWorkerOnly && defaultDeadlockDetectionTimeout == that.defaultDeadlockDetectionTimeout && disableEagerExecution == that.disableEagerExecution - && maxConcurrentEagerActivityExecutionSize == that.maxConcurrentEagerActivityExecutionSize + && maxEagerActivityReservationsPerWorkflowTask + == that.maxEagerActivityReservationsPerWorkflowTask && useBuildIdForVersioning == that.useBuildIdForVersioning && Objects.equals(workerTuner, that.workerTuner) && Objects.equals(maxHeartbeatThrottleInterval, that.maxHeartbeatThrottleInterval) @@ -1137,7 +1140,7 @@ public int hashCode() { defaultHeartbeatThrottleInterval, stickyQueueScheduleToStartTimeout, disableEagerExecution, - maxConcurrentEagerActivityExecutionSize, + maxEagerActivityReservationsPerWorkflowTask, useBuildIdForVersioning, buildId, stickyTaskQueueDrainTimeout, @@ -1189,8 +1192,8 @@ public String toString() { + stickyQueueScheduleToStartTimeout + ", disableEagerExecution=" + disableEagerExecution - + ", maxConcurrentEagerActivityExecutionSize=" - + maxConcurrentEagerActivityExecutionSize + + ", maxEagerActivityReservationsPerWorkflowTask=" + + maxEagerActivityReservationsPerWorkflowTask + ", useBuildIdForVersioning=" + useBuildIdForVersioning + ", buildId='" diff --git a/temporal-sdk/src/test/java/io/temporal/internal/worker/EagerActivitySlotLimiterTest.java b/temporal-sdk/src/test/java/io/temporal/internal/worker/EagerActivitySlotLimiterTest.java deleted file mode 100644 index 41d5a5dea..000000000 --- a/temporal-sdk/src/test/java/io/temporal/internal/worker/EagerActivitySlotLimiterTest.java +++ /dev/null @@ -1,31 +0,0 @@ -package io.temporal.internal.worker; - -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertTrue; - -import org.junit.Test; - -public class EagerActivitySlotLimiterTest { - @Test - public void enforcesLimitUntilReservationIsReleased() { - ActivityWorker.EagerActivitySlotLimiter limiter = - new ActivityWorker.EagerActivitySlotLimiter(2); - - assertTrue(limiter.tryReserve()); - assertTrue(limiter.tryReserve()); - assertFalse(limiter.tryReserve()); - - limiter.release(); - assertTrue(limiter.tryReserve()); - } - - @Test - public void zeroAllowsUnlimitedReservations() { - ActivityWorker.EagerActivitySlotLimiter limiter = - new ActivityWorker.EagerActivitySlotLimiter(0); - - for (int i = 0; i < 1000; i++) { - assertTrue(limiter.tryReserve()); - } - } -} 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 62894f10c..c6765fe5e 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,7 +56,7 @@ public void verifyNewBuilderFromExistingWorkerOptions() { .setDefaultHeartbeatThrottleInterval(Duration.ofSeconds(7)) .setStickyQueueScheduleToStartTimeout(Duration.ofSeconds(60)) .setDisableEagerExecution(false) - .setMaxConcurrentEagerActivityExecutionSize(17) + .setMaxEagerActivityReservationsPerWorkflowTask(17) .setUseBuildIdForVersioning(false) .setBuildId("build-id") .setStickyTaskQueueDrainTimeout(Duration.ofSeconds(15)) @@ -92,8 +93,8 @@ public void verifyNewBuilderFromExistingWorkerOptions() { w1.getStickyQueueScheduleToStartTimeout(), w2.getStickyQueueScheduleToStartTimeout()); assertEquals(w1.isEagerExecutionDisabled(), w2.isEagerExecutionDisabled()); assertEquals( - w1.getMaxConcurrentEagerActivityExecutionSize(), - w2.getMaxConcurrentEagerActivityExecutionSize()); + w1.getMaxEagerActivityReservationsPerWorkflowTask(), + w2.getMaxEagerActivityReservationsPerWorkflowTask()); assertEquals(w1.isUsingBuildIdForVersioning(), w2.isUsingBuildIdForVersioning()); assertEquals(w1.getBuildId(), w2.getBuildId()); assertEquals(w1.getStickyTaskQueueDrainTimeout(), w2.getStickyTaskQueueDrainTimeout()); From 323e49fe5e463e15716570ebb3d523a63eaa255c Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Fri, 24 Jul 2026 13:29:16 -0700 Subject: [PATCH 3/3] Validate eager activity reservation limits --- .../io/temporal/worker/WorkerOptions.java | 8 +- .../io/temporal/worker/WorkerOptionsTest.java | 17 ++++ .../EagerActivityDispatchingTest.java | 87 ++++++++++++++++++- 3 files changed, 108 insertions(+), 4 deletions(-) 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 73bb0ca8f..3346b5314 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java +++ b/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. * - *

The default is 3. Setting this to zero disables eager activity execution. + *

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"); 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 c6765fe5e..1dd61df1d 100644 --- a/temporal-sdk/src/test/java/io/temporal/worker/WorkerOptionsTest.java +++ b/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()); + } + } } 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); + } + } }