activities/execution

180 notes in this subject, read out of this brain and free to use. This is page 2 of 3.

Task Token retrieval for asynchronous completion

To retrieve the Task Token, use the GetInfo() API from the go.temporal.io/sdk/activity package. The Activity calls activity.GetInfo(ctx) and accesses the TaskToken field from the returned ActivityInfo struct. This Task Token must be sent to the external service that will complete the Activity.

Activity return value for asynchronous completion

To indicate that an Activity is completing asynchronously, the Activity Function must return an activity.ErrResultPending error.

CompleteActivity function parameters

The CompleteActivity function takes four parameters: (1) context.Context - context for the operation. (2) taskToken - the binary TaskToken field from the ActivityInfo struct retrieved inside the Activity. (3) result - the return value to record for the Activity, which must match the type of the return value declared by the Activity function. (4) err - the error code to return if the Activity terminates with an error. If err is not null, the value of the result field is ignored.

Example: Get Task Token from Activity context

// Retrieve the Activity information needed to asynchronously complete the Activity. activityInfo := activity.GetInfo(ctx) taskToken := activityInfo.TaskToken // Send the taskToken to the external service that will complete the Activity.

Example: Return pending status from Activity

return "", activity.ErrResultPending

Example: Complete Activity with Temporal Client

// Instantiate a Temporal service client. // The same client can be used to complete or fail any number of Activities. // The client is a heavyweight object that should be created once per process. temporalClient, err := client.Dial(client.Options{}) // Complete the Activity. temporalClient.CompleteActivity(context.Background(), taskToken, result, nil)

Example: Fail an Activity using CompleteActivity

// Fail the Activity. client.CompleteActivity(context.Background(), taskToken, nil, err)

ExecuteActivity API in Go

To spawn an Activity Execution in Go, call ExecuteActivity() inside the Workflow Definition. The API is available from the go.temporal.io/sdk/workflow package. ExecuteActivity() requires an instance of workflow.Context, the Activity function name (as a variable object or string), and any variables to be passed to the Activity Execution. The Activity function name can be provided as a variable object (which allows framework validation of parameters) or as a string. ExecuteActivity returns a Future which can be used to get the result of the Activity Execution.

Activity Execution example in Go

func YourWorkflowDefinition(ctx workflow.Context, param YourWorkflowParam) (*YourWorkflowResultObject, error) { activityOptions := workflow.ActivityOptions{ StartToCloseTimeout: 10 * time.Second, } ctx = workflow.WithActivityOptions(ctx, activityOptions) activityParam := YourActivityParam{ ActivityParamX: param.WorkflowParamX, ActivityParamY: param.WorkflowParamY, } var a *YourActivityObject var activityResult YourActivityResultObject err := workflow.ExecuteActivity(ctx, a.YourActivityDefinition, activityParam).Get(ctx, &activityResult) if err != nil { return nil, err } }

Required Activity Timeouts

The only required value that needs to be set is either a Schedule-To-Close Timeout or a Start-To-Close Timeout. These values are set in the Activity Options. At least one of these two timeouts must be configured.

ActivityOptions fields reference for Go

ActivityOptions fields: | Field | Required | Type | Default | |-------|----------|------|----------| | ActivityID | No | string | None | | TaskQueueName | No | string | Inherits the TaskQueue name from the Workflow | | ScheduleToCloseTimeout | Yes (or StartToCloseTimeout) | time.Duration | ∞ (infinity - no limit) | | ScheduleToStartTimeout | No | time.Duration | ∞ (infinity - no limit) | | StartToCloseTimeout | Yes (or ScheduleToCloseTimeout) | time.Duration | Same as the ScheduleToCloseTimeout | | HeartbeatTimeout | No | time.Duration | (not specified) | | WaitForCancellation | No | bool | false | | OriginalTaskQueueName | No | string | (not specified) | | RetryPolicy | No | RetryPolicy | InitialInterval: 1 second, BackoffCoefficient: 2.0, MaximumInterval: 100 seconds, MaximumAttempts: 0 (unlimited), NonRetryableErrorTypes: empty |

WaitForCancellation Activity Option

If WaitForCancellation is set to true, the Activity Execution will finish executing should there be a Cancellation request. Default is false.

Default RetryPolicy for Activities

The default RetryPolicy for Activities has the following values: InitialInterval of 1 second, BackoffCoefficient of 2.0, MaximumInterval of 100 seconds (100 * InitialInterval), MaximumAttempts of 0 (unlimited retries), and an empty NonRetryableErrorTypes list.

workflow.Future methods for Activity results

ExecuteActivity returns an instance of workflow.Future which has two methods: Get() takes an instance of workflow.Context and a pointer as parameters, populates the variable with the Activity Execution result, and blocks until results are available; IsReady() returns true when the result of the Activity Execution is ready.

Get Activity Execution results in Go

Call the Get() method on the instance of workflow.Future to get the result of the Activity Execution. The type of the result parameter must match the type of the return value declared by the Activity function. The Get() call blocks until the results are available.

Check Activity result readiness before blocking

Use the IsReady() method on workflow.Future first to make sure the Get() call doesn't cause the Workflow Execution to wait on the result. This allows checking if results are available without blocking.

Asynchronous Activity completion definition and purpose

Asynchronous Activity Completion enables the Activity Function to return without the Activity Execution completing. This allows an external system to handle part of the work and then complete the Activity later.

Three steps for asynchronous Activity completion

Step 1: The Activity provides the external system with identifying information needed to complete the Activity Execution, which can be a Task Token or a combination of Namespace, Workflow Id, and Activity Id. Step 2: The Activity Function completes in a way that identifies it as waiting to be completed by an external system. Step 3: The Temporal Client is used to Heartbeat and complete the Activity.

ActivityCompletionClient interface for completing Activities asynchronously

Use the ActivityCompletionClient interface to complete an Activity asynchronously by setting it to the complete() method, passing the task token and result as parameters.

doNotCompleteOnReturn method prevents automatic Activity completion

The doNotCompleteOnReturn() method is called during an Activity Execution to prevent the Activity Execution from completing when its method returns. When this method is called, the Activity Execution does not complete when its method returns and the return value is ignored.

Obtaining task token from Activity execution context

The task token can be obtained by calling Activity.getExecutionContext().getTaskToken(). The task token returns a byte array that can be used as a correlation identifier to complete the Activity asynchronously from an external system.

Asynchronous Activity completion example with Java SDK

Example showing asynchronous Activity completion: ```java @Override public String composeGreeting(String greeting, String name) { // Get the activity execution context ActivityExecutionContext context = Activity.getExecutionContext(); // Set a correlation token that can be used to complete the activity asynchronously byte[] taskToken = context.getTaskToken(); // Execute activity asynchronously using ForkJoinPool ForkJoinPool.commonPool().execute(() -> composeGreetingAsync(taskToken, greeting, name)); context.doNotCompleteOnReturn(); // Since we have set doNotCompleteOnReturn(), the workflow action method return value is ignored. return "ignored"; } // Method that will complete action execution using ActivityCompletionClient private void composeGreetingAsync(byte[] taskToken, String greeting, String name) { String result = greeting + " " + name + "!"; // Complete our workflow activity using ActivityCompletionClient completionClient.complete(taskToken, result); } ```

Identifying information for asynchronous Activity completion

To complete an Activity asynchronously, the Activity must provide identifying information to the external system. This can be either a Task Token or a combination of Namespace, Workflow Id, and Activity Id.

Standalone Activities example with getHandle

ActivityHandle<String> handle = client.getHandle("standalone-activity-id", null, String.class); This example shows how to create a typed handle to a previously started Standalone Activity, passing null as the run ID to target the latest run.

ActivityClient.execute() method for Standalone Activities

Use ActivityClient.execute() to execute a Standalone Activity and block until it completes. This durably enqueues the Standalone Activity in the Temporal Server, waits for it to be executed on a Worker, and then returns the typed result. Call this from application code, not from inside a Workflow Definition. The typed execute() API takes the Activity interface class and an unbound method reference, which the SDK uses to infer the Activity type name and result type at runtime. Alternatively, you can call Activities by string type name.

StartActivityOptions required parameters

StartActivityOptions requires id, taskQueue, and at least one of startToCloseTimeout or scheduleToCloseTimeout.

ActivityClient.start() for non-blocking activity start

Use ActivityClient.start() to start a Standalone Activity and get a handle without waiting for the result. This sends a request to the Temporal Server to durably enqueue the Activity job, without waiting for it to be executed by the Worker. It returns an ActivityHandle that can be used to wait for the result later or manage the Activity.

Getting handles to existing Standalone Activities

Use client.getHandle(activityId, runId, resultType) to create a typed handle to a previously started Standalone Activity. Pass null as the run ID to target the latest run of the given activity ID. You can then use the handle to wait for the result, describe, cancel, or terminate the Activity.

Waiting for Standalone Activity results

Calling client.execute() is equivalent to calling client.start() to durably enqueue the Activity, and then calling handle.getResult() to block until the Activity completes and return the result. To wait asynchronously without blocking the calling thread, use handle.getResultAsync(), which returns a CompletableFuture<R>.

Standalone Activities example with execute

ActivityClient client = ActivityClient.newInstance( service, ActivityClientOptions.newBuilder().setNamespace(profile.getNamespace()).build()); StartActivityOptions options = StartActivityOptions.newBuilder() .setId(ACTIVITY_ID) .setTaskQueue(TASK_QUEUE) .setStartToCloseTimeout(Duration.ofSeconds(10)) .build(); String result = client.execute( GreetingActivities.class, GreetingActivities::composeGreeting, options, "Hello", "World"); System.out.println("Activity result: " + result); This example shows how to execute a Standalone Activity and block until it completes.

Standalone Activities example with start and handle

ActivityHandle<String> handle = client.start( GreetingActivities.class, GreetingActivities::composeGreeting, options, "Hello", "World"); System.out.println("Started activity ID: " + ACTIVITY_ID); // Wait for the result later String result = handle.getResult(); System.out.println("Activity result: " + result); This example shows how to start a Standalone Activity without blocking, and then retrieve the result later using the activity handle.

Activities require ActivityStub and cannot be invoked standalone

Activities are remote procedure calls that must be invoked from within a Workflow using ActivityStub. Activities are not executable on their own. You cannot start an Activity Execution by itself.

Prerequisites before Activity Execution invocation

Before an Activity Execution is invoked, the following must be set: Activity options (either setStartToCloseTimeout or ScheduleToCloseTimeout are required) must be set for the Activity. The Activity must be registered with a Worker. Activity code must be thread-safe.

ActivityStub types: typed and untyped

Activities can be invoked using Workflow.newActivityStub (type-safe) or Workflow.newUntypedActivityStub (untyped). An ActivityStub returns a client-side stub that implements an Activity interface. The untyped Activity stub is useful when the Activity type is not known at compile time, or to invoke Activities implemented in different programming languages.

Activities can be invoked synchronously or asynchronously

Activities can be invoked synchronously (blocking until result is available) or asynchronously (using Async class which returns a Promise).

Async.function and Async.procedure for asynchronous Activity invocation

The Temporal Java SDK provides the Async class with static methods to invoke Activities asynchronously. Use Async.function for Activities that return a result, and Async.procedure for Activities that return void. The calls return a result of type Promise which is similar to Java Future and CompletionStage.

Multiple Activity stubs with independent options

A Workflow can have multiple Activity stubs. Each Activity stub can have its own ActivityOptions defined, allowing different timeout and task queue settings for different Activities.

ActivityExecutionContext provides workflow invocation information

ActivityExecutionContext is a context object passed to each Activity implementation by default, accessible via Activity.getExecutionContext(). It provides getters to access information about the Workflow that invoked the Activity, including namespace, workflowId, runId, activityId, and StartToCloseTimeout. The Activity context information is stored in a thread-local variable, so getExecutionContext() calls succeed only within the thread that invoked the Activity function.

ActivityOptions required timeout configuration

Either a Schedule-To-Close Timeout or a Start-To-Close Timeout must be set in Activity Options. These are the only required values that need to be set for Activity Execution semantics.

ActivityOptions.Builder available timeout options

Available timeouts in ActivityOptions.Builder are: ScheduleToCloseTimeout(), ScheduleToStartTimeout(), and StartToCloseTimeout().

ActivityOptions can be set via ActivityStub or WorkflowImplementationOptions

Activity Options can be set using an ActivityStub within a Workflow implementation, or per-Activity using WorkflowImplementationOptions within a Worker. If options are defined per-Activity Type with WorkflowImplementationOptions.setActivityOptions(), setting them again specifically with ActivityStub in a Workflow will override this setting.

ActivityOptions reference table

ActivityOptions that can be configured for Activity invocation: | Option | Required | Type | |--------|----------|------| | setScheduleToCloseTimeout | Yes (if StartToCloseTimeout is not specified) | Duration | | setScheduleToStartTimeout | No | Duration | | setStartToCloseTimeout | Yes (if ScheduleToCloseTimeout is not specified) | Duration | | setHeartbeatTimeout | No | Duration | | setTaskQueue | No | String | | setRetryOptions | No | RetryOptions | | setCancellationType | No | ActivityCancellationType |

ScheduleToCloseTimeout defaults and behavior

ScheduleToCloseTimeout Type is Duration. Default is Unlimited. Note that if WorkflowRunTimeout and/or WorkflowExecutionTimeout are defined in the Workflow, all Activity retries will stop when either or both of these timeouts are reached.

ScheduleToStartTimeout non-retryable

ScheduleToStartTimeout Type is Duration. Default is Unlimited. This timeout is non-retryable.

StartToCloseTimeout defaults to ScheduleToCloseTimeout

StartToCloseTimeout Type is Duration. Default is ScheduleToCloseTimeout value.

HeartbeatTimeout configuration

HeartbeatTimeout Type is Duration. Default is None.

TaskQueue defaults to Workflow task queue

TaskQueue Type is String. Default is Defaults to the Task Queue that the Workflow was started with.

RetryOptions default is server-defined

RetryOptions Type is RetryOptions. Default is Server-defined Activity Retry policy.

setCancellationType default is TRY_CANCEL

setCancellationType Type is ActivityCancellationType. Default is ActivityCancellationType.TRY_CANCEL.

Activity implementation should be idempotent

A single instance of the Activities implementation is shared across multiple simultaneous Activity invocations. Activity implementation code should be idempotent.

Activity Execution result handling with Promise

The call to spawn an Activity Execution generates the ScheduleActivityTask Command and provides the Workflow with an Awaitable. To get the results of an asynchronously invoked Activity method, use the Promise get method to block until the Activity method result is available.

Asynchronous Activity completion with doNotCompleteOnReturn

Sometimes an Activity Execution lifecycle goes beyond a synchronous method invocation, such as when a request is put in a queue and later a reply comes from a different Worker process. To indicate that an Activity should not be completed upon its method return, call ActivityExecutionContext.doNotCompleteOnReturn() from the original Activity thread. Then later, when replies come, complete the Activity using the ActivityCompletionClient.

Correlating asynchronous Activity completion with TaskToken or IDs

To correlate Activity invocation with asynchronous completion, use either a TaskToken or Workflow and Activity Ids.

ActivityCompletionClient methods for asynchronous completion

When completing an Activity asynchronously using ActivityCompletionClient, use complete(taskToken, result) to complete successfully or completeExceptionally(taskToken, failure) to fail the Activity.

Synchronous Activity invocation example with ActivityStub

Example of synchronous Activity invocation from Workflow: ```java public class FileProcessingWorkflowImpl implements FileProcessingWorkflow { private final FileProcessingActivities activities; public FileProcessingWorkflowImpl() { this.activities = Workflow.newActivityStub( FileProcessingActivities.class, ActivityOptions.newBuilder() .setStartToCloseTimeout(Duration.ofHours(1)) .build()); } @Override public void processFile(Arguments args) { String localName = null; String processedName = null; try { localName = activities.download(args.getSourceBucketName(), args.getSourceFilename()); processedName = activities.processFile(localName); activities.upload(args.getTargetBucketName(), args.getTargetFilename(), processedName); } finally { if (localName != null) { activities.deleteLocalFile(localName); } if (processedName != null) { activities.deleteLocalFile(processedName); } } } } ```

Multiple Activity stubs with different task queues example

Example of Workflow implementation with two typed Activity stubs with different task queues: ```java public FileProcessingWorkflowImpl() { ActivityOptions options1 = ActivityOptions.newBuilder() .setTaskQueue("taskQueue1") .setStartToCloseTimeout(Duration.ofMinutes(10)) .build(); this.store1 = Workflow.newActivityStub(FileProcessingActivities.class, options1); ActivityOptions options2 = ActivityOptions.newBuilder() .setTaskQueue("taskQueue2") .setStartToCloseTimeout(Duration.ofMinutes(5)) .build(); this.store2 = Workflow.newActivityStub(FileProcessingActivities.class, options2); } ```

Untyped Activity stub invocation example

Example of invoking Activities using untyped ActivityStub: ```java // Workflow code ActivityOptions activityOptions = ActivityOptions.newBuilder() .setStartToCloseTimeout(Duration.ofSeconds(3)) .setTaskQueue("simple-queue-node") .build(); ActivityStub activity = Workflow.newUntypedActivityStub(activityOptions); activity.execute("ComposeGreeting", String.class, "Hello World", "Spanish"); ```

Asynchronous Activity invocation with Async.function

Example of asynchronous Activity invocation: ```java Promise<String> localNamePromise = Async.function(activities::download, sourceBucket, sourceFile); ```

Parallel Activity invocations example with Promise.allOf

Example of calling multiple Activity methods in parallel: ```java public void processFile(Arguments args) { List<Promise<String>> localNamePromises = new ArrayList<>(); List<String> processedNames = null; try { // Download all files in parallel. for (String sourceFilename : args.getSourceFilenames()) { Promise<String> localName = Async.function(activities::download, args.getSourceBucketName(), sourceFilename); localNamePromises.add(localName); } List<String> localNames = new ArrayList<>(); for (Promise<String> localName : localNamePromises) { localNames.add(localName.get()); } processedNames = activities.processFiles(localNames); // Upload all results in parallel. List<Promise<Void>> uploadedList = new ArrayList<>(); for (String processedName : processedNames) { Promise<Void> uploaded = Async.procedure( activities::upload, args.getTargetBucketName(), args.getTargetFilename(), processedName); uploadedList.add(uploaded); } // Wait for all uploads to complete. Promise.allOf(uploadedList).get(); } finally { for (Promise<String> localNamePromise : localNamePromises) { // Skip files that haven't completed downloading. if (localNamePromise.isCompleted()) { activities.deleteLocalFile(localNamePromise.get()); } } if (processedNames != null) { for (String processedName : processedNames) { activities.deleteLocalFile(processedName); } } } } ```

ActivityExecutionContext usage example

Example of using ActivityExecutionContext: ```java public class FileProcessingActivitiesImpl implements FileProcessingActivities { @Override public String download(String bucketName, String remoteName, String localName) { ActivityExecutionContext ctx = Activity.getExecutionContext(); ActivityInfo info = ctx.getInfo(); log.info("namespace=" + info.getActivityNamespace()); log.info("workflowId=" + info.getWorkflowId()); log.info("runId=" + info.getRunId()); log.info("activityId=" + info.getActivityId()); log.info("activityTimeout=" + info.getStartToCloseTimeout(); return downloadFileFromS3(bucketName, remoteName, localDirectory + localName); } } ```

Give your agent this brain