From a2a3333f006e30ce8ac2e92aad6b607c9b7d0cef Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Mon, 18 Aug 2025 17:23:28 -0700 Subject: [PATCH 1/7] Update proto submod --- temporal-serviceclient/src/main/proto | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/temporal-serviceclient/src/main/proto b/temporal-serviceclient/src/main/proto index 49f9286fa..c58ba607e 160000 --- a/temporal-serviceclient/src/main/proto +++ b/temporal-serviceclient/src/main/proto @@ -1 +1 @@ -Subproject commit 49f9286fae31a472ba4ca953df6a7432c493085f +Subproject commit c58ba607ec92d35bfd5cce0998742403618383d0 From 4fed033dda2bde4fa54ce99186a42931af48164b Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Mon, 18 Aug 2025 17:38:59 -0700 Subject: [PATCH 2/7] Supress deprecated APIs --- .../failure/DefaultFailureConverter.java | 22 ++++++++++++------- .../internal/nexus/NexusTaskHandlerImpl.java | 1 + 2 files changed, 15 insertions(+), 8 deletions(-) diff --git a/temporal-sdk/src/main/java/io/temporal/failure/DefaultFailureConverter.java b/temporal-sdk/src/main/java/io/temporal/failure/DefaultFailureConverter.java index 79ffc192d..55e190a49 100644 --- a/temporal-sdk/src/main/java/io/temporal/failure/DefaultFailureConverter.java +++ b/temporal-sdk/src/main/java/io/temporal/failure/DefaultFailureConverter.java @@ -166,14 +166,19 @@ private RuntimeException failureToExceptionImpl(Failure failure, DataConverter d case NEXUS_OPERATION_EXECUTION_FAILURE_INFO: { NexusOperationFailureInfo info = failure.getNexusOperationExecutionFailureInfo(); - return new NexusOperationFailure( - failure.getMessage(), - info.getScheduledEventId(), - info.getEndpoint(), - info.getService(), - info.getOperation(), - info.getOperationToken().isEmpty() ? info.getOperationId() : info.getOperationToken(), - cause); + @SuppressWarnings("deprecation") + NexusOperationFailure f = + new NexusOperationFailure( + failure.getMessage(), + info.getScheduledEventId(), + info.getEndpoint(), + info.getService(), + info.getOperation(), + info.getOperationToken().isEmpty() + ? info.getOperationId() + : info.getOperationToken(), + cause); + return f; } case NEXUS_HANDLER_FAILURE_INFO: { @@ -307,6 +312,7 @@ private Failure exceptionToFailure(Throwable throwable) { failure.setCanceledFailureInfo(info); } else if (throwable instanceof NexusOperationFailure) { NexusOperationFailure no = (NexusOperationFailure) throwable; + @SuppressWarnings("deprecation") NexusOperationFailureInfo.Builder op = NexusOperationFailureInfo.newBuilder() .setScheduledEventId(no.getScheduledEventId()) diff --git a/temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusTaskHandlerImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusTaskHandlerImpl.java index 9ba72091a..4b28f6bd7 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusTaskHandlerImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusTaskHandlerImpl.java @@ -186,6 +186,7 @@ private CancelOperationResponse handleCancelledOperation( OperationContext.Builder ctx, CancelOperationRequest task) { ctx.setService(task.getService()).setOperation(task.getOperation()); + @SuppressWarnings("deprecation") // getOperationId kept to support old server for a while OperationCancelDetails operationCancelDetails = OperationCancelDetails.newBuilder() .setOperationToken( From 394b3cb0d2f7c597318a2426ff50b6c6e8bacdf0 Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Mon, 18 Aug 2025 18:02:50 -0700 Subject: [PATCH 3/7] Actually update proto submodule --- temporal-serviceclient/src/main/proto | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/temporal-serviceclient/src/main/proto b/temporal-serviceclient/src/main/proto index c58ba607e..d96bd55e8 160000 --- a/temporal-serviceclient/src/main/proto +++ b/temporal-serviceclient/src/main/proto @@ -1 +1 @@ -Subproject commit c58ba607ec92d35bfd5cce0998742403618383d0 +Subproject commit d96bd55e87799e9f6a33a1c40a56cfa932566bdf From c2a13470b7e2e3689d41b8c5438b00201ecb84dd Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Mon, 18 Aug 2025 18:09:12 -0700 Subject: [PATCH 4/7] Add fairness keys/weights --- .../java/io/temporal/common/Priority.java | 70 +++++++++++++++-- .../internal/common/ProtoConverters.java | 21 ++++- .../temporal/workflow/PriorityInfoTest.java | 76 ++++++++++--------- 3 files changed, 122 insertions(+), 45 deletions(-) diff --git a/temporal-sdk/src/main/java/io/temporal/common/Priority.java b/temporal-sdk/src/main/java/io/temporal/common/Priority.java index 62dde7c0c..eceffec3a 100644 --- a/temporal-sdk/src/main/java/io/temporal/common/Priority.java +++ b/temporal-sdk/src/main/java/io/temporal/common/Priority.java @@ -31,12 +31,16 @@ public static Priority getDefaultInstance() { public static final class Builder { private int priorityKey; + private String fairnessKey; + private float fairnessWeight; private Builder(Priority options) { if (options == null) { return; } this.priorityKey = options.getPriorityKey(); + this.fairnessKey = options.getFairnessKey(); + this.fairnessWeight = options.getFairnessWeight(); } /** @@ -55,16 +59,44 @@ public Builder setPriorityKey(int priorityKey) { return this; } + /** + * A fairness key is a short string used for balancing task dispatch. Tasks with the same + * fairness key will be processed proportionally according to their fairness weight. + * + *

If not set, inherits from the parent workflow or uses an empty string if there is no + * parent. + */ + public Builder setFairnessKey(String fairnessKey) { + this.fairnessKey = fairnessKey; + return this; + } + + /** + * A fairness weight determines the relative proportion of task processing for a given fairness + * key. The weight should be a positive number. A higher weight means more tasks will be + * processed for that fairness key. + * + *

If not set or 0, defaults to 1.0. If there is a parent workflow, inherits from the parent. + */ + public Builder setFairnessWeight(float fairnessWeight) { + this.fairnessWeight = fairnessWeight; + return this; + } + public Priority build() { - return new Priority(priorityKey); + return new Priority(priorityKey, fairnessKey, fairnessWeight); } } - private Priority(int priorityKey) { + private Priority(int priorityKey, String fairnessKey, float fairnessWeight) { this.priorityKey = priorityKey; + this.fairnessKey = fairnessKey; + this.fairnessWeight = fairnessWeight; } private final int priorityKey; + private final String fairnessKey; + private final float fairnessWeight; /** * See {@link Builder#setPriorityKey(int)} @@ -75,20 +107,48 @@ public int getPriorityKey() { return priorityKey; } + /** + * See {@link Builder#setFairnessKey(String)} + * + * @return The fairness key + */ + public String getFairnessKey() { + return fairnessKey; + } + + /** + * See {@link Builder#setFairnessWeight(float)} + * + * @return The fairness weight + */ + public float getFairnessWeight() { + return fairnessWeight; + } + @Override public String toString() { - return "Priority{" + "priorityKey=" + priorityKey + '}'; + return "Priority{" + + "priorityKey=" + + priorityKey + + ", fairnessKey='" + + fairnessKey + + '\'' + + ", fairnessWeight=" + + fairnessWeight + + '}'; } @Override public boolean equals(Object o) { if (o == null || getClass() != o.getClass()) return false; Priority priority = (Priority) o; - return priorityKey == priority.priorityKey; + return priorityKey == priority.priorityKey + && Float.compare(priority.fairnessWeight, fairnessWeight) == 0 + && Objects.equals(fairnessKey, priority.fairnessKey); } @Override public int hashCode() { - return Objects.hashCode(priorityKey); + return Objects.hash(priorityKey, fairnessKey, fairnessWeight); } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/common/ProtoConverters.java b/temporal-sdk/src/main/java/io/temporal/internal/common/ProtoConverters.java index 9895528aa..0981adb0b 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/common/ProtoConverters.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/common/ProtoConverters.java @@ -8,14 +8,27 @@ public class ProtoConverters { public static Priority toProto(io.temporal.common.Priority priority) { - return Priority.newBuilder().setPriorityKey(priority.getPriorityKey()).build(); + Priority.Builder builder = Priority.newBuilder().setPriorityKey(priority.getPriorityKey()); + if (priority.getFairnessKey() != null) { + builder.setFairnessKey(priority.getFairnessKey()); + } + if (priority.getFairnessWeight() != 0.0f) { + builder.setFairnessWeight(priority.getFairnessWeight()); + } + return builder.build(); } @Nonnull public static io.temporal.common.Priority fromProto(@Nonnull Priority priority) { - return io.temporal.common.Priority.newBuilder() - .setPriorityKey(priority.getPriorityKey()) - .build(); + io.temporal.common.Priority.Builder builder = + io.temporal.common.Priority.newBuilder().setPriorityKey(priority.getPriorityKey()); + if (!priority.getFairnessKey().isEmpty()) { + builder.setFairnessKey(priority.getFairnessKey()); + } + if (priority.getFairnessWeight() != 0.0f) { + builder.setFairnessWeight(priority.getFairnessWeight()); + } + return builder.build(); } public static io.temporal.api.deployment.v1.WorkerDeploymentVersion toProto( diff --git a/temporal-sdk/src/test/java/io/temporal/workflow/PriorityInfoTest.java b/temporal-sdk/src/test/java/io/temporal/workflow/PriorityInfoTest.java index a444f5c4a..339c5f103 100644 --- a/temporal-sdk/src/test/java/io/temporal/workflow/PriorityInfoTest.java +++ b/temporal-sdk/src/test/java/io/temporal/workflow/PriorityInfoTest.java @@ -33,10 +33,15 @@ public void testPriority() { TestWorkflow1.class, WorkflowOptions.newBuilder() .setTaskQueue(testWorkflowRule.getTaskQueue()) - .setPriority(Priority.newBuilder().setPriorityKey(5).build()) + .setPriority( + Priority.newBuilder() + .setPriorityKey(5) + .setFairnessKey("tenant-123") + .setFairnessWeight(2.5f) + .build()) .build()); String result = workflowStub.execute(testWorkflowRule.getTaskQueue()); - assertEquals("5", result); + assertEquals("5:tenant-123:2.5", result); } @ActivityInterface @@ -47,15 +52,18 @@ public interface PriorityActivities { public static class PriorityActivitiesImpl implements PriorityActivities { @Override public String activity1(String a1) { - return String.valueOf( - Activity.getExecutionContext().getInfo().getPriority().getPriorityKey()); + Priority priority = Activity.getExecutionContext().getInfo().getPriority(); + String key = priority.getFairnessKey() != null ? priority.getFairnessKey() : "null"; + return priority.getPriorityKey() + ":" + key + ":" + priority.getFairnessWeight(); } } public static class TestPriorityChildWorkflow implements TestWorkflows.TestWorkflowReturnString { @Override public String execute() { - return String.valueOf(Workflow.getInfo().getPriority().getPriorityKey()); + Priority priority = Workflow.getInfo().getPriority(); + String key = priority.getFairnessKey() != null ? priority.getFairnessKey() : "null"; + return priority.getPriorityKey() + ":" + key + ":" + priority.getFairnessWeight(); } } @@ -70,12 +78,17 @@ public String execute(String taskQueue) { ActivityOptions.newBuilder() .setTaskQueue(taskQueue) .setStartToCloseTimeout(Duration.ofSeconds(10)) - .setPriority(Priority.newBuilder().setPriorityKey(3).build()) + .setPriority( + Priority.newBuilder() + .setPriorityKey(3) + .setFairnessKey("override") + .setFairnessWeight(1.5f) + .build()) .setDisableEagerExecution(true) .build()) .activity1("1"); - Assert.assertEquals("3", priority); - // Test that of if no priority is set the workflows priority is used + Assert.assertEquals("3:override:1.5", priority); + // Test that if no priority is set the workflow's priority is used priority = Workflow.newActivityStub( PriorityActivities.class, @@ -85,46 +98,37 @@ public String execute(String taskQueue) { .setDisableEagerExecution(true) .build()) .activity1("2"); - Assert.assertEquals("5", priority); - // Test that of if a default priority is set the workflows priority is used - priority = - Workflow.newActivityStub( - PriorityActivities.class, - ActivityOptions.newBuilder() - .setTaskQueue(taskQueue) - .setStartToCloseTimeout(Duration.ofSeconds(10)) - .setPriority(Priority.newBuilder().build()) - .setDisableEagerExecution(true) - .build()) - .activity1("2"); - Assert.assertEquals("5", priority); + Assert.assertEquals("5:tenant-123:2.5", priority); // Test that the priority is passed to child workflows priority = Workflow.newChildWorkflowStub( TestWorkflows.TestWorkflowReturnString.class, ChildWorkflowOptions.newBuilder() - .setPriority(Priority.newBuilder().setPriorityKey(1).build()) + .setPriority( + Priority.newBuilder() + .setPriorityKey(1) + .setFairnessKey("child") + .setFairnessWeight(0.5f) + .build()) .build()) .execute(); - Assert.assertEquals("1", priority); - // Test that of no priority is set the workflows priority is used + Assert.assertEquals("1:child:0.5", priority); + // Test that if no priority is set the workflow's priority is used priority = Workflow.newChildWorkflowStub( TestWorkflows.TestWorkflowReturnString.class, ChildWorkflowOptions.newBuilder().build()) .execute(); - Assert.assertEquals("5", priority); - // Test that if a default priority is set the workflows priority is used - priority = - Workflow.newChildWorkflowStub( - TestWorkflows.TestWorkflowReturnString.class, - ChildWorkflowOptions.newBuilder() - .setPriority(Priority.newBuilder().build()) - .build()) - .execute(); - Assert.assertEquals("5", priority); - // Return the workflows priority - return String.valueOf(Workflow.getInfo().getPriority().getPriorityKey()); + Assert.assertEquals("5:tenant-123:2.5", priority); + // Return the workflow's priority + Priority workflowPriority = Workflow.getInfo().getPriority(); + String key = + workflowPriority.getFairnessKey() != null ? workflowPriority.getFairnessKey() : "null"; + return workflowPriority.getPriorityKey() + + ":" + + key + + ":" + + workflowPriority.getFairnessWeight(); } } } From 1e567b5e53be7449b10c898801c771997fda9546 Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Mon, 18 Aug 2025 23:06:40 -0700 Subject: [PATCH 5/7] Fix test server merge --- .../io/temporal/internal/testservice/StateMachines.java | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/temporal-test-server/src/main/java/io/temporal/internal/testservice/StateMachines.java b/temporal-test-server/src/main/java/io/temporal/internal/testservice/StateMachines.java index f1138f9b4..242fb552e 100644 --- a/temporal-test-server/src/main/java/io/temporal/internal/testservice/StateMachines.java +++ b/temporal-test-server/src/main/java/io/temporal/internal/testservice/StateMachines.java @@ -2563,9 +2563,17 @@ static Priority mergePriorities(Priority parent, Priority child) { } Priority.Builder result = Priority.newBuilder(); result.setPriorityKey(parent.getPriorityKey()); + result.setFairnessKey(child.getFairnessKey()); + result.setFairnessWeight(child.getFairnessWeight()); if (child.getPriorityKey() != 0) { result.setPriorityKey(child.getPriorityKey()); } + if (!child.getFairnessKey().isEmpty()) { + result.setFairnessKey(child.getFairnessKey()); + } + if (child.getFairnessWeight() != 0) { + result.setFairnessWeight(child.getFairnessWeight()); + } return result.build(); } } From 00ea7d8cc11fdab4d65cab039abe63a37ac84c53 Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Mon, 18 Aug 2025 23:12:12 -0700 Subject: [PATCH 6/7] Fix docstrings --- .../java/io/temporal/common/Priority.java | 27 +++++++++++++------ 1 file changed, 19 insertions(+), 8 deletions(-) diff --git a/temporal-sdk/src/main/java/io/temporal/common/Priority.java b/temporal-sdk/src/main/java/io/temporal/common/Priority.java index eceffec3a..3432d4431 100644 --- a/temporal-sdk/src/main/java/io/temporal/common/Priority.java +++ b/temporal-sdk/src/main/java/io/temporal/common/Priority.java @@ -60,11 +60,16 @@ public Builder setPriorityKey(int priorityKey) { } /** - * A fairness key is a short string used for balancing task dispatch. Tasks with the same - * fairness key will be processed proportionally according to their fairness weight. + * FairnessKey is a short string that's used as a key for a fairness balancing mechanism. It may + * correspond to a tenant id, or to a fixed string like "high" or "low". The default is the + * empty string. * - *

If not set, inherits from the parent workflow or uses an empty string if there is no - * parent. + *

>The fairness mechanism attempts to dispatch tasks for a given key in proportion to its + * weight. For example, using a thousand distinct tenant ids, each with a weight of 1.0 (the + * default) will result in each tenant getting a roughly equal share of task dispatch + * throughput. + * + *

Fairness keys are limited to 64 bytes. */ public Builder setFairnessKey(String fairnessKey) { this.fairnessKey = fairnessKey; @@ -72,11 +77,17 @@ public Builder setFairnessKey(String fairnessKey) { } /** - * A fairness weight determines the relative proportion of task processing for a given fairness - * key. The weight should be a positive number. A higher weight means more tasks will be - * processed for that fairness key. + * FairnessWeight for a task can come from multiple sources for flexibility. From highest to + * lowest precedence: + * + *

    + *
  • Weights for a small set of keys can be overridden in task queue configuration with an + * API. + *
  • It can be attached to the workflow/activity in this field. + *
  • The default weight of 1.0 will be used. + *
* - *

If not set or 0, defaults to 1.0. If there is a parent workflow, inherits from the parent. + *

Weight values are clamped to the range [0.001, 1000]. */ public Builder setFairnessWeight(float fairnessWeight) { this.fairnessWeight = fairnessWeight; From 5196dbc12e26d4d471a9cc4590b290fa90e971e1 Mon Sep 17 00:00:00 2001 From: Spencer Judge Date: Tue, 19 Aug 2025 08:58:10 -0700 Subject: [PATCH 7/7] Use CLI prerelease --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 93a0d2144..aad276fd2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -72,7 +72,7 @@ jobs: - name: Start containerized server and dependencies env: - TEMPORAL_CLI_VERSION: 1.4.0 + TEMPORAL_CLI_VERSION: 1.4.1-cloud-v1-29-0-139-2.0 run: | wget -O temporal_cli.tar.gz https://github.com/temporalio/cli/releases/download/v${TEMPORAL_CLI_VERSION}/temporal_cli_${TEMPORAL_CLI_VERSION}_linux_amd64.tar.gz tar -xzf temporal_cli.tar.gz