Skip to content

Commit 23590a1

Browse files
authored
AWS Lambda (Java) (#2901)
* Lambda worker support for Java SDK * Support dynamic worker and activity registrations in Lambda * Behavioral fixes * Corrections and completions * Add test injection comments * Make LambdaWorkerOptions immutable * Add per-invocation Lambda worker configuration Add LambdaWorker overloads that accept an invocation configurator so handlers can adjust worker options using the AWS Lambda Context before Temporal service stubs, clients, and workers are created. Copy the base options for each invocation, allow required fields such as task queue to be supplied dynamically, and keep invocation-local shutdown hooks from leaking across warm invocations. Run those hooks if the per-invocation configurator or final option validation fails. Document the new overload and cover the lifecycle behavior with unit tests. * Factor out OpenTelemetry into separate contrib package * Review comments * Make OTel implementation a plugin * No magic constants! * Rename `LambdaWorker.run()` -> `LambdaWorker.define()`
1 parent f86c766 commit 23590a1

24 files changed

Lines changed: 5422 additions & 0 deletions
Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
# Temporal AWS Lambda worker module
2+
3+
This module provides a direct AWS Lambda Java handler for running a Temporal worker for one Lambda invocation.
4+
5+
## Usage
6+
7+
Add `temporal-aws-lambda` next to your Temporal SDK dependency, then expose the returned handler from your Lambda class:
8+
9+
```java
10+
import com.amazonaws.services.lambda.runtime.Context;
11+
import com.amazonaws.services.lambda.runtime.RequestHandler;
12+
import io.temporal.aws.lambda.LambdaWorker;
13+
import io.temporal.common.WorkerDeploymentVersion;
14+
15+
public final class Handler implements RequestHandler<Object, Void> {
16+
private static final RequestHandler<Object, Void> WORKER =
17+
LambdaWorker.define(
18+
new WorkerDeploymentVersion("orders-worker", "2026-06-02"),
19+
builder ->
20+
builder
21+
.setTaskQueue("orders")
22+
.registerWorkflowImplementationTypes(OrderWorkflowImpl.class)
23+
.registerActivitiesImplementations(new OrderActivitiesImpl()));
24+
25+
@Override
26+
public Void handleRequest(Object input, Context context) {
27+
return WORKER.handleRequest(input, context);
28+
}
29+
}
30+
```
31+
32+
`TEMPORAL_TASK_QUEUE` can provide the task queue. If it is not set, call `setTaskQueue`.
33+
34+
Connection options are loaded with `temporal-envconfig` when the handler is constructed during Lambda cold start. The Lambda worker checks `TEMPORAL_CONFIG_FILE` first, then `$LAMBDA_TASK_ROOT/temporal.toml`, then `./temporal.toml`, then falls back to the envconfig defaults and Temporal environment variables. The `configure` callback, passed as the second parameter to `LambdaWorker.define`, also runs during handler construction, so non-invocation configuration is prepared once and reused.
35+
36+
Use the per-invocation configuration overload when final options depend on the Lambda `Context` or on resources opened for one invocation. The cold-start `configure` callback passed as the second parameter to `LambdaWorker.define` still runs once; the invocation callback runs before Temporal service stubs, client, and worker are created:
37+
38+
```java
39+
private static final RequestHandler<Object, Void> WORKER =
40+
LambdaWorker.define(
41+
new WorkerDeploymentVersion("orders-worker", "2026-06-02"),
42+
builder -> builder.registerWorkflowImplementationTypes(OrderWorkflowImpl.class),
43+
(builder, context) -> {
44+
builder.setTaskQueue(taskQueueFor(context));
45+
builder.addShutdownHook(() -> cleanupInvocationResources(context));
46+
});
47+
```
48+
49+
Each invocation receives a fresh copy of the base options, so mutations and shutdown hooks added by the invocation callback do not leak across warm invocations. If the invocation callback throws after adding shutdown hooks, those hooks still run for cleanup. Reusable resources such as AWS SDK clients should usually be created during cold start and reused across warm invocations.
50+
51+
If you need to assemble options outside the `define` callback, call `LambdaWorkerOptions.newBuilderFromEnvironment()`, configure the returned builder, call `build()`, and pass the options to `LambdaWorker.newHandler(...)`.
52+
53+
Dynamic workflow and activity implementations can be registered with `registerDynamicWorkflowImplementationType(...)` and `registerDynamicActivityImplementation(...)`. Java SDK worker rules still apply: only one dynamic workflow implementation type and one dynamic activity implementation can be registered per worker.
54+
55+
The handler creates one worker per invocation, starts the worker, shuts it down before the Lambda deadline, runs shutdown hooks in order, and closes service stubs. Worker deployment versioning is always enabled for the supplied `WorkerDeploymentVersion`. If neither client nor worker identity is set by the user, each invocation uses `<awsRequestId>@<invokedFunctionArn>` as the Temporal identity.
56+
57+
`shutdownDeadlineBuffer` is the full shutdown window reserved at the end of the Lambda invocation. The default is 7 seconds: 5 seconds for `gracefulShutdownTimeout` and 2 seconds for hooks and service stubs. The worker runs until `remainingTime - shutdownDeadlineBuffer`, then stops and awaits termination for `gracefulShutdownTimeout`. If you change `gracefulShutdownTimeout` without explicitly setting `shutdownDeadlineBuffer`, the buffer is recomputed as `gracefulShutdownTimeout + 2s`.
58+
If you explicitly set `shutdownDeadlineBuffer`, it must be greater than or equal to `gracefulShutdownTimeout`.
59+
60+
## OpenTelemetry
61+
62+
`OtelLambdaWorkerConfigurationHelper.configure(builder)` is a Lambda-specific facade over `temporal-opentelemetry`. It creates an `OpenTelemetryPlugin`, configures it with AWS X-Ray-compatible trace ID generation, installs it on service stubs options so it propagates to the workflow client and worker factory, and registers the plugin's timed flush hook for each Lambda invocation. The hook reports buffered Tally values before the OpenTelemetry provider hook force-flushes exporters. To enable it, call the helper from the handler initializer:
63+
64+
```java
65+
private static final RequestHandler<Object, Void> WORKER =
66+
LambdaWorker.define(
67+
new WorkerDeploymentVersion("orders-worker", "2026-06-02"),
68+
builder -> {
69+
OtelLambdaWorkerConfigurationHelper.configure(builder);
70+
builder
71+
.setTaskQueue("orders")
72+
.registerWorkflowImplementationTypes(OrderWorkflowImpl.class);
73+
});
74+
```
75+
76+
The helper defaults the OTLP endpoint from `OTEL_EXPORTER_OTLP_ENDPOINT`, then `http://localhost:4317`. It defaults the service name from `OTEL_SERVICE_NAME`, then `AWS_LAMBDA_FUNCTION_NAME`, then `temporal-lambda-worker`, and sets it on the OpenTelemetry resource. To use an application-owned provider, call `builder.setOpenTelemetry(...)`; in that path, no exporters are created and the helper only installs the plugin and per-invocation flush hook. Providers and scopes are not closed after each invocation.
77+
78+
Use `OtelLambdaWorkerConfigurationHelper.configureMetrics(...)`, `OtelLambdaWorkerConfigurationHelper.configureTracing(...)`, and `OtelLambdaWorkerConfigurationHelper.configureFlushHook(...)` when you want to compose metrics, tracing, or provider flushing separately around an application-owned OpenTelemetry instance. Use `temporal-opentelemetry` directly for non-Lambda serverless adapters or long-running workers.
79+
80+
For Java logging, this module depends on `slf4j-api` only. It does not bundle a runtime logging binding, so Lambda log formatting remains owned by the application.
Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
description = '''Temporal Java SDK AWS Lambda Worker Support Module'''
2+
3+
ext {
4+
awsLambdaJavaCoreVersion = '1.4.0'
5+
otelVersion = '1.25.0'
6+
}
7+
8+
dependencies {
9+
api platform("io.opentelemetry:opentelemetry-bom:$otelVersion")
10+
11+
// This module shouldn't carry temporal-sdk with it, especially for situations when users may
12+
// be using a shaded artifact.
13+
compileOnly project(':temporal-sdk')
14+
compileOnly "javax.annotation:javax.annotation-api:$annotationApiVersion"
15+
16+
api "com.amazonaws:aws-lambda-java-core:$awsLambdaJavaCoreVersion"
17+
api project(':temporal-opentelemetry')
18+
19+
implementation project(':temporal-envconfig')
20+
api "io.opentelemetry:opentelemetry-api"
21+
implementation "io.opentelemetry.contrib:opentelemetry-aws-xray:$otelVersion"
22+
implementation "org.slf4j:slf4j-api:$slf4jVersion"
23+
24+
testImplementation project(':temporal-sdk')
25+
testImplementation "io.opentelemetry:opentelemetry-sdk"
26+
testImplementation "junit:junit:${junitVersion}"
27+
28+
testRuntimeOnly group: 'ch.qos.logback', name: 'logback-classic', version: "${logbackVersion}"
29+
}
Lines changed: 149 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,149 @@
1+
package io.temporal.aws.lambda;
2+
3+
import io.temporal.client.WorkflowClient;
4+
import io.temporal.client.WorkflowClientOptions;
5+
import io.temporal.common.converter.EncodedValues;
6+
import io.temporal.serviceclient.WorkflowServiceStubs;
7+
import io.temporal.serviceclient.WorkflowServiceStubsOptions;
8+
import io.temporal.worker.Worker;
9+
import io.temporal.worker.WorkerFactory;
10+
import io.temporal.worker.WorkerFactoryOptions;
11+
import io.temporal.worker.WorkerOptions;
12+
import io.temporal.worker.WorkflowImplementationOptions;
13+
import io.temporal.workflow.Functions;
14+
import java.time.Duration;
15+
import java.util.concurrent.TimeUnit;
16+
17+
final class DefaultLambdaWorkerRuntime implements LambdaWorkerRuntime {
18+
@Override
19+
public Invocation create(
20+
WorkflowServiceStubsOptions serviceStubsOptions,
21+
WorkflowClientOptions clientOptions,
22+
WorkerFactoryOptions workerFactoryOptions,
23+
String taskQueue,
24+
WorkerOptions workerOptions) {
25+
WorkflowServiceStubs stubs = WorkflowServiceStubs.newServiceStubs(serviceStubsOptions);
26+
WorkerFactory factory = null;
27+
try {
28+
WorkflowClient client = WorkflowClient.newInstance(stubs, clientOptions);
29+
factory = WorkerFactory.newInstance(client, workerFactoryOptions);
30+
Worker worker = factory.newWorker(taskQueue, workerOptions);
31+
return new DefaultInvocation(stubs, factory, worker);
32+
} catch (RuntimeException e) {
33+
if (factory != null) {
34+
try {
35+
factory.shutdownNow();
36+
} catch (RuntimeException shutdownException) {
37+
e.addSuppressed(shutdownException);
38+
}
39+
}
40+
try {
41+
stubs.shutdownNow();
42+
} catch (RuntimeException shutdownException) {
43+
e.addSuppressed(shutdownException);
44+
}
45+
throw e;
46+
}
47+
}
48+
49+
private static final class DefaultInvocation implements Invocation {
50+
private final WorkflowServiceStubs stubs;
51+
private final WorkerFactory factory;
52+
private final WorkerRegistrar registrar;
53+
54+
private DefaultInvocation(WorkflowServiceStubs stubs, WorkerFactory factory, Worker worker) {
55+
this.stubs = stubs;
56+
this.factory = factory;
57+
this.registrar = new DefaultWorkerRegistrar(worker);
58+
}
59+
60+
@Override
61+
public WorkerRegistrar getWorkerRegistrar() {
62+
return registrar;
63+
}
64+
65+
@Override
66+
public void start() {
67+
factory.start();
68+
}
69+
70+
@Override
71+
public void shutdown() {
72+
factory.shutdown();
73+
}
74+
75+
@Override
76+
public void shutdownNow() {
77+
factory.shutdownNow();
78+
}
79+
80+
@Override
81+
public void awaitTermination(Duration timeout) {
82+
factory.awaitTermination(timeout.toMillis(), TimeUnit.MILLISECONDS);
83+
}
84+
85+
@Override
86+
public boolean isTerminated() {
87+
return factory.isTerminated();
88+
}
89+
90+
@Override
91+
public void closeStubs(Duration timeout) {
92+
stubs.shutdown();
93+
if (!stubs.awaitTermination(timeout.toMillis(), TimeUnit.MILLISECONDS)) {
94+
stubs.shutdownNow();
95+
}
96+
}
97+
}
98+
99+
private static final class DefaultWorkerRegistrar implements WorkerRegistrar {
100+
private final Worker worker;
101+
102+
private DefaultWorkerRegistrar(Worker worker) {
103+
this.worker = worker;
104+
}
105+
106+
@Override
107+
public void registerWorkflowImplementationTypes(Class<?>... workflowImplementationClasses) {
108+
worker.registerWorkflowImplementationTypes(workflowImplementationClasses);
109+
}
110+
111+
@Override
112+
public void registerWorkflowImplementationTypes(
113+
WorkflowImplementationOptions options, Class<?>... workflowImplementationClasses) {
114+
worker.registerWorkflowImplementationTypes(options, workflowImplementationClasses);
115+
}
116+
117+
@Override
118+
public <R> void registerWorkflowImplementationFactory(
119+
Class<R> workflowInterface, Functions.Func<R> factory) {
120+
worker.registerWorkflowImplementationFactory(workflowInterface, factory);
121+
}
122+
123+
@Override
124+
public <R> void registerWorkflowImplementationFactory(
125+
Class<R> workflowInterface,
126+
Functions.Func1<EncodedValues, R> factory,
127+
WorkflowImplementationOptions options) {
128+
worker.registerWorkflowImplementationFactory(workflowInterface, factory, options);
129+
}
130+
131+
@Override
132+
public <R> void registerWorkflowImplementationFactory(
133+
Class<R> workflowInterface,
134+
Functions.Func<R> factory,
135+
WorkflowImplementationOptions options) {
136+
worker.registerWorkflowImplementationFactory(workflowInterface, factory, options);
137+
}
138+
139+
@Override
140+
public void registerActivitiesImplementations(Object... activityImplementations) {
141+
worker.registerActivitiesImplementations(activityImplementations);
142+
}
143+
144+
@Override
145+
public void registerNexusServiceImplementation(Object... nexusServiceImplementations) {
146+
worker.registerNexusServiceImplementation(nexusServiceImplementations);
147+
}
148+
}
149+
}

0 commit comments

Comments
 (0)