mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-30 14:56:19 +00:00
d8af9436cc
## What changed? This PR adds a new `GetTaskQueueUserData` RPC to the admin service. Given a namespace, task queue name, task queue type, and optional partition ID (default is 0, which is `root`), it returns the user data currently loaded by that partition. This PR wraps the existing `GetTaskQueueUserData` RPC in Matching Service. We will also create a `tdbg` command which calls this Admin Service RPC, in a separate PR. ## Why? Each task queue family has associated metadata, stored in [TaskQueueUserData](https://github.com/temporalio/temporal/blob/main/proto/internal/temporal/server/api/persistence/v1/task_queues.proto). Metadata related to worker versioning, queue rate limiting, fairness are all stored in `TaskQueueUserData`. TaskQueueUserData is replicated from the root partition to all other partitions. However, there is no admin-accessible way to read the user data loaded by a specific partition or compare versions across partitions to diagnose replication lag. ## Files changed | File | Change | |---|---| | `proto/internal/.../adminservice/v1/request_response.proto` | Added `GetTaskQueueUserDataRequest` and `GetTaskQueueUserDataResponse` messages | | `proto/internal/.../adminservice/v1/service.proto` | Added `GetTaskQueueUserData` RPC to `AdminService` | | `service/frontend/admin_handler.go` | Implemented `AdminHandler.GetTaskQueueUserData`: validates request, resolves namespace → ID, builds partition RPC name via `tqid`, calls matching service, returns per-type entry + version | | `service/frontend/admin_handler_test.go` | Added unit tests | ## How did you test it? - [x] built - [x] run locally and tested manually - [x] added new unit test(s) - [x] added new integration test(s) - not applicable, not touching persistence layer - [x] added new functional test(s) ### Unit tests 100% unit test coverage | Test case | Input | Expected | |---|---|---| | Nil request | `request == nil` | `errRequestNotSet` | | Empty namespace | `namespace == ""` | `errNamespaceNotSet` | | Namespace not found | Namespace registry returns not-found | Error propagated; matching never called | | Invalid task queue name | `task_queue` starts with `/_sys/` | `INVALID_ARGUMENT` from `tqid.NewTaskQueueFamily`; matching never called | | Root partition | `partition_id=0`, workflow type | Sends bare name `my-queue` to matching; returns correct `user_data` and `version` | | Non-root partition | `partition_id=1`, workflow type | Sends mangled name `/_sys/my-queue/1` to matching | | No per-type data | Matching returns response with empty `per_type` map | `user_data` is nil; `version` still populated | | Matching error | Matching client returns error | Error propagated to caller | ### Functional tests | Test | Setup | What it verifies | |------------------------------------------------|--------------------------------|----------------------------------------------------------------------------------| | TestAdminGetTaskQueueUserData_RootPartition | Write fairness weight config to a workflow task queue | Admin RPC resolves namespace by name, routes to root partition (partition_id=0), returns version > 0 and non-nil per-type data | | TestAdminGetTaskQueueUserData_NonRootPartition | Same write, then poll until non-root partition replicates | Admin RPC routes to a non-root partition (partition_id=1) via mangled name, returns the same version as root after replication | ### Manual tests <details> <summary>Setup</summary> 1. Build and start the server: `make temporal-server && make start-sqlite` 2. Create namespace: `temporal operator namespace create default` 3. Insert assignment rule: `temporal task-queue versioning insert-assignment-rule` </details> <details> <summary>Case 1 — Root partition, workflow type</summary> ``` grpcurl -plaintext \ -d '{"namespace":"default","task_queue":"my-queue","task_queue_type":"TASK_QUEUE_TYPE_WORKFLOW"}' \ localhost:7233 \ temporal.server.api.adminservice.v1.AdminService/GetTaskQueueUserData ``` ```json { "version": "1" } ``` </details> <details> <summary>Case 2 — Root partition, activity type</summary> ``` grpcurl -plaintext \ -d '{"namespace":"default","task_queue":"my-queue","task_queue_type":"TASK_QUEUE_TYPE_ACTIVITY"}' \ localhost:7233 \ temporal.server.api.adminservice.v1.AdminService/GetTaskQueueUserData ``` ```json { "version": "1" } ``` </details> <details> <summary>Case 3 — Non-root partition, workflow type</summary> ``` grpcurl -plaintext \ -d '{"namespace":"default","task_queue":"my-queue","task_queue_type":"TASK_QUEUE_TYPE_WORKFLOW","partition_id":1}' \ localhost:7233 \ temporal.server.api.adminservice.v1.AdminService/GetTaskQueueUserData ``` ```json { "version": "1" } ``` </details> <details> <summary>Case 4 — Non-root partition, activity type</summary> ``` grpcurl -plaintext \ -d '{"namespace":"default","task_queue":"my-queue","task_queue_type":"TASK_QUEUE_TYPE_ACTIVITY","partition_id":1}' \ localhost:7233 \ temporal.server.api.adminservice.v1.AdminService/GetTaskQueueUserData ``` ```json { "version": "1" } ``` </details> <details> <summary>Case 5 — Non-root partition, activity type, with user data</summary> Setup: Add rate limit config ``` grpcurl -plaintext \ -d '{ "namespace": "default", "task_queue": "my-queue", "task_queue_type": "TASK_QUEUE_TYPE_ACTIVITY", "update_queue_rate_limit": { "rate_limit": { "requests_per_second": 50.0 }, "reason": "manual test" } }' \ localhost:7233 \ temporal.api.workflowservice.v1.WorkflowService/UpdateTaskQueueConfig ``` ``` grpcurl -plaintext \ -d '{ "namespace": "default", "task_queue": "my-queue", "task_queue_type": "TASK_QUEUE_TYPE_ACTIVITY" }' \ localhost:7233 \ temporal.server.api.adminservice.v1.AdminService/GetTaskQueueUserData ``` ```json { "userData": { "config": { "queueRateLimit": { "rateLimit": { "requestsPerSecond": 50 }, "metadata": { "reason": "manual test", "updateTime": "2026-04-13T21:40:39.888Z" } } } }, "version": "2" } ``` </details> <details> <summary>Case 6 — Namespace not found</summary> ``` grpcurl -plaintext \ -d '{"namespace":"nonexistent","task_queue":"my-queue","task_queue_type":"TASK_QUEUE_TYPE_WORKFLOW"}' \ localhost:7233 \ temporal.server.api.adminservice.v1.AdminService/GetTaskQueueUserData ``` ``` ERROR: Code: NotFound Message: Namespace nonexistent is not found. ``` `NOT_FOUND` from namespace registry; matching never called. </details> --------- Co-authored-by: Hanzo Dev <[email protected]>
688 lines
21 KiB
Go
688 lines
21 KiB
Go
// Code generated by cmd/tools/genrpcwrappers. DO NOT EDIT.
|
|
|
|
package admin
|
|
|
|
import (
|
|
"context"
|
|
|
|
"go.temporal.io/server/api/adminservice/v1"
|
|
"google.golang.org/grpc"
|
|
|
|
"go.temporal.io/server/common/backoff"
|
|
)
|
|
|
|
func (c *retryableClient) AddOrUpdateRemoteCluster(
|
|
ctx context.Context,
|
|
request *adminservice.AddOrUpdateRemoteClusterRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.AddOrUpdateRemoteClusterResponse, error) {
|
|
var resp *adminservice.AddOrUpdateRemoteClusterResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.AddOrUpdateRemoteCluster(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) AddSearchAttributes(
|
|
ctx context.Context,
|
|
request *adminservice.AddSearchAttributesRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.AddSearchAttributesResponse, error) {
|
|
var resp *adminservice.AddSearchAttributesResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.AddSearchAttributes(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) AddTasks(
|
|
ctx context.Context,
|
|
request *adminservice.AddTasksRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.AddTasksResponse, error) {
|
|
var resp *adminservice.AddTasksResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.AddTasks(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) CancelDLQJob(
|
|
ctx context.Context,
|
|
request *adminservice.CancelDLQJobRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.CancelDLQJobResponse, error) {
|
|
var resp *adminservice.CancelDLQJobResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.CancelDLQJob(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) CloseShard(
|
|
ctx context.Context,
|
|
request *adminservice.CloseShardRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.CloseShardResponse, error) {
|
|
var resp *adminservice.CloseShardResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.CloseShard(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) DeepHealthCheck(
|
|
ctx context.Context,
|
|
request *adminservice.DeepHealthCheckRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.DeepHealthCheckResponse, error) {
|
|
var resp *adminservice.DeepHealthCheckResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.DeepHealthCheck(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) DeleteWorkflowExecution(
|
|
ctx context.Context,
|
|
request *adminservice.DeleteWorkflowExecutionRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.DeleteWorkflowExecutionResponse, error) {
|
|
var resp *adminservice.DeleteWorkflowExecutionResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.DeleteWorkflowExecution(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) DescribeCluster(
|
|
ctx context.Context,
|
|
request *adminservice.DescribeClusterRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.DescribeClusterResponse, error) {
|
|
var resp *adminservice.DescribeClusterResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.DescribeCluster(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) DescribeDLQJob(
|
|
ctx context.Context,
|
|
request *adminservice.DescribeDLQJobRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.DescribeDLQJobResponse, error) {
|
|
var resp *adminservice.DescribeDLQJobResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.DescribeDLQJob(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) DescribeHistoryHost(
|
|
ctx context.Context,
|
|
request *adminservice.DescribeHistoryHostRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.DescribeHistoryHostResponse, error) {
|
|
var resp *adminservice.DescribeHistoryHostResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.DescribeHistoryHost(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) DescribeMutableState(
|
|
ctx context.Context,
|
|
request *adminservice.DescribeMutableStateRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.DescribeMutableStateResponse, error) {
|
|
var resp *adminservice.DescribeMutableStateResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.DescribeMutableState(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) DescribeTaskQueuePartition(
|
|
ctx context.Context,
|
|
request *adminservice.DescribeTaskQueuePartitionRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.DescribeTaskQueuePartitionResponse, error) {
|
|
var resp *adminservice.DescribeTaskQueuePartitionResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.DescribeTaskQueuePartition(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) ForceUnloadTaskQueuePartition(
|
|
ctx context.Context,
|
|
request *adminservice.ForceUnloadTaskQueuePartitionRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.ForceUnloadTaskQueuePartitionResponse, error) {
|
|
var resp *adminservice.ForceUnloadTaskQueuePartitionResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.ForceUnloadTaskQueuePartition(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) GenerateLastHistoryReplicationTasks(
|
|
ctx context.Context,
|
|
request *adminservice.GenerateLastHistoryReplicationTasksRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.GenerateLastHistoryReplicationTasksResponse, error) {
|
|
var resp *adminservice.GenerateLastHistoryReplicationTasksResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.GenerateLastHistoryReplicationTasks(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) GetDLQMessages(
|
|
ctx context.Context,
|
|
request *adminservice.GetDLQMessagesRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.GetDLQMessagesResponse, error) {
|
|
var resp *adminservice.GetDLQMessagesResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.GetDLQMessages(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) GetDLQReplicationMessages(
|
|
ctx context.Context,
|
|
request *adminservice.GetDLQReplicationMessagesRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.GetDLQReplicationMessagesResponse, error) {
|
|
var resp *adminservice.GetDLQReplicationMessagesResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.GetDLQReplicationMessages(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) GetDLQTasks(
|
|
ctx context.Context,
|
|
request *adminservice.GetDLQTasksRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.GetDLQTasksResponse, error) {
|
|
var resp *adminservice.GetDLQTasksResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.GetDLQTasks(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) GetNamespace(
|
|
ctx context.Context,
|
|
request *adminservice.GetNamespaceRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.GetNamespaceResponse, error) {
|
|
var resp *adminservice.GetNamespaceResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.GetNamespace(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) GetNamespaceReplicationMessages(
|
|
ctx context.Context,
|
|
request *adminservice.GetNamespaceReplicationMessagesRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.GetNamespaceReplicationMessagesResponse, error) {
|
|
var resp *adminservice.GetNamespaceReplicationMessagesResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.GetNamespaceReplicationMessages(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) GetReplicationMessages(
|
|
ctx context.Context,
|
|
request *adminservice.GetReplicationMessagesRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.GetReplicationMessagesResponse, error) {
|
|
var resp *adminservice.GetReplicationMessagesResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.GetReplicationMessages(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) GetSearchAttributes(
|
|
ctx context.Context,
|
|
request *adminservice.GetSearchAttributesRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.GetSearchAttributesResponse, error) {
|
|
var resp *adminservice.GetSearchAttributesResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.GetSearchAttributes(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) GetShard(
|
|
ctx context.Context,
|
|
request *adminservice.GetShardRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.GetShardResponse, error) {
|
|
var resp *adminservice.GetShardResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.GetShard(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) GetTaskQueueTasks(
|
|
ctx context.Context,
|
|
request *adminservice.GetTaskQueueTasksRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.GetTaskQueueTasksResponse, error) {
|
|
var resp *adminservice.GetTaskQueueTasksResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.GetTaskQueueTasks(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) GetTaskQueueUserData(
|
|
ctx context.Context,
|
|
request *adminservice.GetTaskQueueUserDataRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.GetTaskQueueUserDataResponse, error) {
|
|
var resp *adminservice.GetTaskQueueUserDataResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.GetTaskQueueUserData(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) GetWorkflowExecutionRawHistory(
|
|
ctx context.Context,
|
|
request *adminservice.GetWorkflowExecutionRawHistoryRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.GetWorkflowExecutionRawHistoryResponse, error) {
|
|
var resp *adminservice.GetWorkflowExecutionRawHistoryResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.GetWorkflowExecutionRawHistory(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) GetWorkflowExecutionRawHistoryV2(
|
|
ctx context.Context,
|
|
request *adminservice.GetWorkflowExecutionRawHistoryV2Request,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.GetWorkflowExecutionRawHistoryV2Response, error) {
|
|
var resp *adminservice.GetWorkflowExecutionRawHistoryV2Response
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.GetWorkflowExecutionRawHistoryV2(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) ImportWorkflowExecution(
|
|
ctx context.Context,
|
|
request *adminservice.ImportWorkflowExecutionRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.ImportWorkflowExecutionResponse, error) {
|
|
var resp *adminservice.ImportWorkflowExecutionResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.ImportWorkflowExecution(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) ListClusterMembers(
|
|
ctx context.Context,
|
|
request *adminservice.ListClusterMembersRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.ListClusterMembersResponse, error) {
|
|
var resp *adminservice.ListClusterMembersResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.ListClusterMembers(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) ListClusters(
|
|
ctx context.Context,
|
|
request *adminservice.ListClustersRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.ListClustersResponse, error) {
|
|
var resp *adminservice.ListClustersResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.ListClusters(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) ListHistoryTasks(
|
|
ctx context.Context,
|
|
request *adminservice.ListHistoryTasksRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.ListHistoryTasksResponse, error) {
|
|
var resp *adminservice.ListHistoryTasksResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.ListHistoryTasks(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) ListQueues(
|
|
ctx context.Context,
|
|
request *adminservice.ListQueuesRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.ListQueuesResponse, error) {
|
|
var resp *adminservice.ListQueuesResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.ListQueues(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) MergeDLQMessages(
|
|
ctx context.Context,
|
|
request *adminservice.MergeDLQMessagesRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.MergeDLQMessagesResponse, error) {
|
|
var resp *adminservice.MergeDLQMessagesResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.MergeDLQMessages(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) MergeDLQTasks(
|
|
ctx context.Context,
|
|
request *adminservice.MergeDLQTasksRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.MergeDLQTasksResponse, error) {
|
|
var resp *adminservice.MergeDLQTasksResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.MergeDLQTasks(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) MigrateSchedule(
|
|
ctx context.Context,
|
|
request *adminservice.MigrateScheduleRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.MigrateScheduleResponse, error) {
|
|
var resp *adminservice.MigrateScheduleResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.MigrateSchedule(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) PurgeDLQMessages(
|
|
ctx context.Context,
|
|
request *adminservice.PurgeDLQMessagesRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.PurgeDLQMessagesResponse, error) {
|
|
var resp *adminservice.PurgeDLQMessagesResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.PurgeDLQMessages(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) PurgeDLQTasks(
|
|
ctx context.Context,
|
|
request *adminservice.PurgeDLQTasksRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.PurgeDLQTasksResponse, error) {
|
|
var resp *adminservice.PurgeDLQTasksResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.PurgeDLQTasks(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) ReapplyEvents(
|
|
ctx context.Context,
|
|
request *adminservice.ReapplyEventsRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.ReapplyEventsResponse, error) {
|
|
var resp *adminservice.ReapplyEventsResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.ReapplyEvents(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) RebuildMutableState(
|
|
ctx context.Context,
|
|
request *adminservice.RebuildMutableStateRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.RebuildMutableStateResponse, error) {
|
|
var resp *adminservice.RebuildMutableStateResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.RebuildMutableState(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) RefreshWorkflowTasks(
|
|
ctx context.Context,
|
|
request *adminservice.RefreshWorkflowTasksRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.RefreshWorkflowTasksResponse, error) {
|
|
var resp *adminservice.RefreshWorkflowTasksResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.RefreshWorkflowTasks(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) RemoveRemoteCluster(
|
|
ctx context.Context,
|
|
request *adminservice.RemoveRemoteClusterRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.RemoveRemoteClusterResponse, error) {
|
|
var resp *adminservice.RemoveRemoteClusterResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.RemoveRemoteCluster(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) RemoveSearchAttributes(
|
|
ctx context.Context,
|
|
request *adminservice.RemoveSearchAttributesRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.RemoveSearchAttributesResponse, error) {
|
|
var resp *adminservice.RemoveSearchAttributesResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.RemoveSearchAttributes(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) RemoveTask(
|
|
ctx context.Context,
|
|
request *adminservice.RemoveTaskRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.RemoveTaskResponse, error) {
|
|
var resp *adminservice.RemoveTaskResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.RemoveTask(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) ResendReplicationTasks(
|
|
ctx context.Context,
|
|
request *adminservice.ResendReplicationTasksRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.ResendReplicationTasksResponse, error) {
|
|
var resp *adminservice.ResendReplicationTasksResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.ResendReplicationTasks(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) StartAdminBatchOperation(
|
|
ctx context.Context,
|
|
request *adminservice.StartAdminBatchOperationRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.StartAdminBatchOperationResponse, error) {
|
|
var resp *adminservice.StartAdminBatchOperationResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.StartAdminBatchOperation(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|
|
|
|
func (c *retryableClient) SyncWorkflowState(
|
|
ctx context.Context,
|
|
request *adminservice.SyncWorkflowStateRequest,
|
|
opts ...grpc.CallOption,
|
|
) (*adminservice.SyncWorkflowStateResponse, error) {
|
|
var resp *adminservice.SyncWorkflowStateResponse
|
|
op := func(ctx context.Context) error {
|
|
var err error
|
|
resp, err = c.client.SyncWorkflowState(ctx, request, opts...)
|
|
return err
|
|
}
|
|
err := backoff.ThrottleRetryContext(ctx, op, c.policy, c.isRetryable)
|
|
return resp, err
|
|
}
|