Workflow or activity randomly not found

Community Support
1

I created a simple Grpc Go application that triggers and registers a workflow and a go service that registers an activity for that workflow. Somehow, it randomly fail the workflow due to errors like:

  • unable to find workflow type: SingUpWorkflow. Supported types: [] in the activity worker
  • unable to find activityType=createCompanyActivity. Supported types: [] in the workflow worker

I am not sure to fully understand what this is happening and how to resolve this

I have Temporal (version 1.22.4) running and deployed via HelmChat.
The SDK version are

go.temporal.io/api v1.24.0
go.temporal.io/sdk v1.25.1

The grpc workflow service:
I can create a client and register the workflow (main):

tempo, err := client.Dial(client.Options{
  HostPort:  "temporal-frontend.event:7233",
  Namespace: "default",
})
if err != nil {
  l.Fatal(ctx, err.Error())
}

// grpc service that will trigger the workflow
srv := newGrpcCloudApi(tempo)
if err != nil {
  l.Fatal(ctx, err.Error())
}


w := worker.New(tempo, "account-signup", worker.Options{})
w.RegisterWorkflow(SingUpWorkflow)
err = w.Start()
if err != nil {
  l.Fatal(ctx, err.Error())
}

My workflow:

func SingUpWorkflow(ctx workflow.Context, request SignUpRequest) (string, error) {

	retrypolicy := &temporal.RetryPolicy{
		MaximumAttempts: 1,
	}

	options := workflow.ActivityOptions{
		TaskQueue:           "account-signup",
		StartToCloseTimeout: 5 * time.Second,
		RetryPolicy:         retrypolicy,
	}

	ctx = workflow.WithActivityOptions(ctx, options)

	createCompanyReq := pbCompany.CreateCompanyRequest{Name: request.Name}
	createCompanyResp := pbCompany.CreateCompanyResponse{}
	err := workflow.ExecuteActivity(ctx, "createCompanyActivity", &createCompanyReq).Get(ctx, &createCompanyResp)
	if err != nil {
		return "", err
	}

	return createCompanyResp.Name, nil
}

Grpc Function for triggering the workflow

func (g *grpcCloudAPI) Signup(ctx context.Context, request *pb.SignupRequest) (*pb.SignupResponse, error) {

options := client.StartWorkflowOptions{
  ID:                       fmt.Sprintf("%s-%s", "account-signup", id.New()),
  TaskQueue:                "account-signup",
  WorkflowExecutionTimeout: 20 * time.Second,
}

signup := SignUpRequest{
  Name:     request.Name,
}

we, err := g.workflowClient.ExecuteWorkflow(ctx, options, SingUpWorkflow, signup)
if err != nil {
  return nil, err
}

response := ""
err = we.Get(ctx, &response)
if err != nil {
	return nil, err
}

return &pb.SignupResponse{
  Name:      response,
  }, nil
}

The activity service

tempo, err := client.Dial(client.Options{
  HostPort:  "temporal-frontend.event:7233",
  Namespace: "default",
})
if err != nil {
  l.Fatal(ctx, err.Error())
}

worker := newWorker(tempo)
err = worker.Start()
  if err != nil {
  l.Fatal(ctx, err.Error())
}
defer worker.Stop()

Activity

type temporalWorker struct {
	w   worker.Worker
}

func newWorker(client client.Client) *temporalWorker {
	w := worker.New(client, "account-signup", worker.Options{})
	return &temporalWorker{
		w:   w,
	}
}

func (wt *temporalWorker) Start() error {

	// register all activities
	wt.w.RegisterActivity(wt.createCompanyActivity)
	return wt.w.Start()
}

func (wt *temporalWorker) Stop() {
	wt.w.Stop()
}

func (wt *temporalWorker) createCompanyActivity(ctx context.Context, request *pb.CreateCompanyRequest) (*pb.CreateCompanyResponse, error) {


	return &pb.CreateCompanyResponse{
		Id:   "ABC",
		Name: "ABC",
	}, nil
}

Workflow:

Screenshot 2024-01-21 at 2.32.14 PM
Screenshot 2024-01-21 at 2.32.14 PM822Ă—1733 145 KB

2

You have two processes that listen on the same task queue name, but each process supports only workflow or activity. Temporal delivers tasks to all processes that listen to a given task queue. So, the process that hosts workflow only can receive activity tasks, and the process that hosts only activities receives workflow tasks. The solution is to use different task queue names for different types of processes.

Change your activity worker to listen on “account-signup-activities” and specify this task queue name in activity.Options used to invoke activities inside the workflow.

3

Hi Maxim,

I see my mistake now, thank you sooo much for your kind reply!

1 Like