Add Poller Group fields (#744)

- **Add poller groups**
- **rename resource_id to routing key for nexus**
- **fix typo**
- **Add poller group fields**

<!-- Describe what has changed in this PR -->
**What changed?**
Adding `poller-group-id` and `poller-group-info` fields to the poll
APIs.

<!-- Tell your future self why have you made these changes -->
**Why?**
SDK needs to pass poller group ID in some APIs according to server
instructions.

<!-- Are there any breaking changes on binary or code level? -->
**Breaking changes**
None

<!-- If this breaks the Server, please provide the Server PR to merge
right after this PR was merged. -->
**Server PR**
No new API added.

Author: ShahabT

openapi/openapiv2.json

@@ -15379,6 +15379,30 @@
         "pollerScalingDecision": {
           "$ref": "#/definitions/v1PollerScalingDecision",
           "description": "Server-advised information the SDK may use to adjust its poller count."
+        },
+        "pollerGroupId": {
+          "type": "string",
+          "description": "This poller group ID identifies the owner of the workflow task awaiting for query response.\nCorresponding RespondQueryTaskCompleted should pass this value for proper routing."
+        },
+        "pollerGroupInfos": {
+          "type": "array",
+          "items": {
+            "type": "object",
+            "$ref": "#/definitions/v1PollerGroupInfo"
+          },
+          "description": "The weighted list of poller groups IDs that client should use for future polls to this task\nqueue. Client is expected to:\n  1. Maintain minimum number of pollers no less than the number of groups.\n  2. Try to assign the next poll to a group without any pending polls,\n  3. If every group has some pending polls, assign the next poll to a group randomly\n    according to the weights."
+        }
+      }
+    },
+    "v1PollerGroupInfo": {
+      "type": "object",
+      "properties": {
+        "id": {
+          "type": "string"
+        },
+        "weight": {
+          "type": "number",
+          "format": "float"
         }
       }
     },

openapi/openapiv3.yaml

@@ -12220,6 +12220,30 @@ components:
           allOf:
             - $ref: '#/components/schemas/PollerScalingDecision'
           description: Server-advised information the SDK may use to adjust its poller count.
+        pollerGroupId:
+          type: string
+          description: |-
+            This poller group ID identifies the owner of the workflow task awaiting for query response.
+             Corresponding RespondQueryTaskCompleted should pass this value for proper routing.
+        pollerGroupInfos:
+          type: array
+          items:
+            $ref: '#/components/schemas/PollerGroupInfo'
+          description: |-
+            The weighted list of poller groups IDs that client should use for future polls to this task
+             queue. Client is expected to:
+               1. Maintain minimum number of pollers no less than the number of groups.
+               2. Try to assign the next poll to a group without any pending polls,
+               3. If every group has some pending polls, assign the next poll to a group randomly
+                 according to the weights.
+    PollerGroupInfo:
+      type: object
+      properties:
+        id:
+          type: string
+        weight:
+          type: number
+          format: float
     PollerInfo:
       type: object
       properties:

temporal/api/taskqueue/v1/message.proto

@@ -303,6 +303,11 @@ message TimestampedCompatibleBuildIdRedirectRule {
     google.protobuf.Timestamp create_time = 2;
 }
 
+message PollerGroupInfo {
+  string id = 1;
+  float weight = 2;
+}
+
 // Attached to task responses to give hints to the SDK about how it may adjust its number of
 // pollers.
 message PollerScalingDecision {

temporal/api/workflowservice/v1/request_response.proto

@@ -260,6 +260,11 @@ message GetWorkflowExecutionHistoryReverseResponse {
 message PollWorkflowTaskQueueRequest {
     string namespace = 1;
     temporal.api.taskqueue.v1.TaskQueue task_queue = 2;
+    // Unless this is the first poll, the client must pass one of the poller group IDs received in
+    // `poller_group_infos` of the last the PollWorkflowTaskQueueResponse according to the
+    // instructions. If not set, the poll is routed randomly which can cause it being blocked
+    // without receiving a task while the queue actually has tasks in another server location.
+    string poller_group_id = 10;
     // The identity of the worker/client who is polling this task queue
     string identity = 3;
     // A unique key for this worker instance, used for tracking worker lifecycle.
@@ -282,7 +287,7 @@ message PollWorkflowTaskQueueRequest {
     temporal.api.deployment.v1.WorkerDeploymentOptions deployment_options = 6;
 
     // Removed in 1.55.0; was temporal.api.worker.v1.WorkerHeartbeat worker_heartbeat
-    reserved 7; 
+    reserved 7;
     reserved "worker_heartbeat";
 }
 
@@ -338,6 +343,16 @@ message PollWorkflowTaskQueueResponse {
     repeated temporal.api.protocol.v1.Message messages = 15;
     // Server-advised information the SDK may use to adjust its poller count.
     temporal.api.taskqueue.v1.PollerScalingDecision poller_scaling_decision = 16;
+    // This poller group ID identifies the owner of the workflow task awaiting for query response.
+    // Corresponding RespondQueryTaskCompleted should pass this value for proper routing.
+    string poller_group_id = 17;
+    // The weighted list of poller groups IDs that client should use for future polls to this task
+    // queue. Client is expected to:
+    //   1. Maintain minimum number of pollers no less than the number of groups.
+    //   2. Try to assign the next poll to a group without any pending polls,
+    //   3. If every group has some pending polls, assign the next poll to a group randomly
+    //     according to the weights.
+    repeated temporal.api.taskqueue.v1.PollerGroupInfo poller_group_infos = 18;
 }
 
 message RespondWorkflowTaskCompletedRequest {
@@ -458,6 +473,11 @@ message RespondWorkflowTaskFailedResponse {
 message PollActivityTaskQueueRequest {
     string namespace = 1;
     temporal.api.taskqueue.v1.TaskQueue task_queue = 2;
+    // Unless this is the first poll, the client must pass one of the poller group IDs received in
+    // `poller_group_infos` of the last the PollActivityTaskQueueResponse according to the
+    // instructions. If not set, the poll is routed randomly which can cause it being blocked
+    // without receiving a task while the queue actually has tasks in another server location.
+    string poller_group_id = 10;
     // The identity of the worker/client
     string identity = 3;
     // A unique key for this worker instance, used for tracking worker lifecycle.
@@ -536,6 +556,13 @@ message PollActivityTaskQueueResponse {
     temporal.api.common.v1.Priority priority = 19;
     // The run ID of the activity execution, only set for standalone activities.
     string activity_run_id = 20;
+    // The weighted list of poller groups IDs that client should use for future polls to this task
+    // queue. Client is expected to:
+    //   1. Maintain minimum number of pollers no less than the number of groups.
+    //   2. Try to assign the next poll to a group without any pending polls,
+    //   3. If every group has some pending polls, assign the next poll to a group randomly
+    //     according to the weights.
+    repeated temporal.api.taskqueue.v1.PollerGroupInfo poller_group_infos = 21;
 }
 
 message RecordActivityTaskHeartbeatRequest {
@@ -1058,6 +1085,9 @@ message RespondQueryTaskCompletedRequest {
     // Why did the task fail? It's important to note that many of the variants in this enum cannot
     // apply to worker responses. See the type's doc for more.
     temporal.api.enums.v1.WorkflowTaskFailedCause cause = 8;
+    // Client must forward the poller_group_id received in PollWorkflowTaskQueueResponse for proper
+    // routing of the response.
+    string poller_group_id = 9;
 }
 
 message RespondQueryTaskCompletedResponse {
@@ -1907,12 +1937,17 @@ message PollWorkflowExecutionUpdateResponse {
 
 message PollNexusTaskQueueRequest {
     string namespace = 1;
+    temporal.api.taskqueue.v1.TaskQueue task_queue = 3;
+    // Unless this is the first poll, the client must pass one of the poller group IDs received in
+    // `poller_group_infos` of the last the PollNexusTaskQueueResponse according to the
+    // instructions. If not set, the poll is routed randomly which can cause it being blocked
+    // without receiving a task while the queue actually has tasks in another server location.
+    string poller_group_id = 9;
     // The identity of the client who initiated this request.
     string identity = 2;
     // A unique key for this worker instance, used for tracking worker lifecycle.
     // This is guaranteed to be unique, whereas identity is not guaranteed to be unique.
     string worker_instance_key = 8;
-    temporal.api.taskqueue.v1.TaskQueue task_queue = 3;
     // Information about this worker's build identifier and if it is choosing to use the versioning
     // feature. See the `WorkerVersionCapabilities` docstring for more.
     // Deprecated. Replaced by deployment_options.
@@ -1931,6 +1966,18 @@ message PollNexusTaskQueueResponse {
     temporal.api.nexus.v1.Request request = 2;
     // Server-advised information the SDK may use to adjust its poller count.
     temporal.api.taskqueue.v1.PollerScalingDecision poller_scaling_decision = 3;
+    // This poller group ID identifies the owner of the nexus task awaiting for synchronous
+    // response.
+    // Corresponding `RespondNexusTaskCompleted` and `RespondNexusTaskFailed` calls should pass this
+    // value for proper response routing.
+    string poller_group_id = 4;
+    // The weighted list of poller groups IDs that client should use for future polls to this task
+    // queue. Client is expected to:
+    //   1. Maintain minimum number of pollers no less than the number of groups.
+    //   2. Try to assign the next poll to a group without any pending polls,
+    //   3. If every group has some pending polls, assign the next poll to a group randomly
+    //     according to the weights.
+    repeated temporal.api.taskqueue.v1.PollerGroupInfo poller_group_infos = 5;
 }
 
 message RespondNexusTaskCompletedRequest {
@@ -1941,6 +1988,9 @@ message RespondNexusTaskCompletedRequest {
     bytes task_token = 3;
     // Embedded response to be translated into a frontend response.
     temporal.api.nexus.v1.Response response = 4;
+    // Client must forward the poller_group_id received in PollNexusTaskQueueResponse for proper
+    // routing of the response.
+    string poller_group_id = 5;
 }
 
 message RespondNexusTaskCompletedResponse {
@@ -1956,6 +2006,9 @@ message RespondNexusTaskFailedRequest {
     temporal.api.nexus.v1.HandlerError error = 4 [deprecated = true];
     // The error the handler failed with. Must contain a NexusHandlerFailureInfo object.
     temporal.api.failure.v1.Failure failure = 5;
+    // Client must forward the poller_group_id received in PollNexusTaskQueueResponse for proper
+    // routing of the response.
+    string poller_group_id = 6;
 }
 
 message RespondNexusTaskFailedResponse {

temporal/api/workflowservice/v1/service.proto

@@ -166,6 +166,10 @@ service WorkflowService {
     // (-- api-linter: core::0127::http-annotation=disabled
     //     aip.dev/not-precedent: We do not expose worker API to HTTP. --)
     rpc PollWorkflowTaskQueue (PollWorkflowTaskQueueRequest) returns (PollWorkflowTaskQueueResponse) {
+      option (temporal.api.protometa.v1.request_header) = {
+        header: "temporal-resource-id"
+        value: "poller:{poller_group_id}"
+      };
     }
 
     // RespondWorkflowTaskCompleted is called by workers to successfully complete workflow tasks
@@ -219,6 +223,10 @@ service WorkflowService {
     // (-- api-linter: core::0127::http-annotation=disabled
     //     aip.dev/not-precedent: We do not expose worker API to HTTP. --)
     rpc PollActivityTaskQueue (PollActivityTaskQueueRequest) returns (PollActivityTaskQueueResponse) {
+      option (temporal.api.protometa.v1.request_header) = {
+        header: "temporal-resource-id"
+        value: "poller:{poller_group_id}"
+      };
     }
 
     // RecordActivityTaskHeartbeat is optionally called by workers while they execute activities.
@@ -618,7 +626,12 @@ service WorkflowService {
     //
     // (-- api-linter: core::0127::http-annotation=disabled
     //     aip.dev/not-precedent: We do not expose worker API to HTTP. --)
-    rpc RespondQueryTaskCompleted (RespondQueryTaskCompletedRequest) returns (RespondQueryTaskCompletedResponse) {}
+    rpc RespondQueryTaskCompleted (RespondQueryTaskCompletedRequest) returns (RespondQueryTaskCompletedResponse) {
+      option (temporal.api.protometa.v1.request_header) = {
+        header: "temporal-resource-id"
+        value: "poller:{poller_group_id}"
+      };
+    }
 
     // ResetStickyTaskQueue resets the sticky task queue related information in the mutable state of
     // a given workflow. This is prudent for workers to perform if a workflow has been paged out of
@@ -1296,18 +1309,30 @@ service WorkflowService {
     // (-- api-linter: core::0127::http-annotation=disabled
     //     aip.dev/not-precedent: We do not expose worker API to HTTP. --)
     rpc PollNexusTaskQueue(PollNexusTaskQueueRequest) returns (PollNexusTaskQueueResponse) {
+      option (temporal.api.protometa.v1.request_header) = {
+        header: "temporal-resource-id"
+        value: "poller:{poller_group_id}"
+      };
     }
 
     // RespondNexusTaskCompleted is called by workers to respond to Nexus tasks received via PollNexusTaskQueue.
     // (-- api-linter: core::0127::http-annotation=disabled
     //     aip.dev/not-precedent: We do not expose worker API to HTTP. --)
     rpc RespondNexusTaskCompleted(RespondNexusTaskCompletedRequest) returns (RespondNexusTaskCompletedResponse) {
+      option (temporal.api.protometa.v1.request_header) = {
+        header: "temporal-resource-id"
+        value: "poller:{poller_group_id}"
+      };
     }
 
     // RespondNexusTaskFailed is called by workers to fail Nexus tasks received via PollNexusTaskQueue.
     // (-- api-linter: core::0127::http-annotation=disabled
     //     aip.dev/not-precedent: We do not expose worker API to HTTP. --)
     rpc RespondNexusTaskFailed(RespondNexusTaskFailedRequest) returns (RespondNexusTaskFailedResponse) {
+      option (temporal.api.protometa.v1.request_header) = {
+        header: "temporal-resource-id"
+        value: "poller:{poller_group_id}"
+      };
     }
 
     // UpdateActivityOptions is called by the client to update the options of an activity by its ID or type.