a valid extracted backend context anchors the execution trace (this is what makes the durable spans share one
@@ -34,4 +36,15 @@ public interface ContextExtractor {
* @return the extracted context, or {@code null} if no context is available
*/
ExtractedContext extract();
+
+ /**
+ * Extracts context using the invocation and its immutable X-Ray header snapshot. Existing extractors retain their
+ * no-argument behavior; implementations that need invocation-local carriers can override this method.
+ *
+ * A null snapshot denotes an unavailable runtime accessor and permits legacy fallback. An empty string denotes
+ * an available accessor with no invocation header; the built-in extractor treats it as authoritative absence.
+ */
+ default ExtractedContext extract(InvocationInfo info) {
+ return extract();
+ }
}
diff --git a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/ExecutionOtelPlugin.java b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/ExecutionOtelPlugin.java
index b23bb19835..617bab5bb5 100644
--- a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/ExecutionOtelPlugin.java
+++ b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/ExecutionOtelPlugin.java
@@ -21,11 +21,13 @@
import java.time.Instant;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import software.amazon.lambda.durable.plugin.DurableExecutionPlugin;
-import software.amazon.lambda.durable.plugin.ExclusivePluginGroup;
+import software.amazon.lambda.durable.plugin.DurableExecutionPluginFactory;
+import software.amazon.lambda.durable.plugin.HandlerScoped;
import software.amazon.lambda.durable.plugin.InvocationEndInfo;
import software.amazon.lambda.durable.plugin.InvocationInfo;
import software.amazon.lambda.durable.plugin.OperationEndInfo;
@@ -63,10 +65,21 @@
*
*
The Workflow and Invocation spans share one execution trace, anchored at the execution ancestor resolved at
* invocation start: a valid propagated remote server span becomes that ancestor directly, otherwise a synthetic
- * execution root anchors the trace. The trace ID is stable across invocations of the same execution. When using
- * {@link #ExecutionOtelPlugin()}, the plugin resolves the global provider at invocation start. If the OpenTelemetry
- * Java agent is not initialized yet, telemetry is disabled for that entire invocation and provider resolution is
- * retried on the next invocation.
+ * execution root anchors the trace. The trace ID is stable across invocations of the same execution, because it is
+ * derived from the execution ARN and start time rather than carried in the plugin.
+ *
+ *
Lifetime. One instance serves exactly one Lambda invocation: {@link #factory()} and its overloads return a
+ * {@link DurableExecutionPluginFactory} that the SDK calls once per invocation. Invocation-level references are
+ * released at return; an unfinished operation retains its matching cleanup until its owner unwinds. Everything about
+ * the invocation — the execution ARN, the resolved execution trace and ancestor, the sampling intent, the Invocation
+ * span, the deferred Workflow span context — is therefore a {@code final} field, resolved in the constructor from the
+ * {@link InvocationInfo} the factory receives. Nothing is reset between invocations because nothing is carried between
+ * them.
+ *
+ *
What belongs to the execution environment stays in the factory's {@link OtelPluginEnvironment}: the configuration,
+ * the ID generator, and either the application-owned tracer provider (built once) or the lazily resolved ADOT global
+ * provider. An invocation whose instance cannot resolve the global provider emits no telemetry at all, and the next
+ * invocation's instance resolves it again.
*
*
Status mapping (parity with the Python/JS references):
*
@@ -86,47 +99,63 @@
* current, so {@code Span.current()} enrichment is not recorded on the final operation span. The placeholder uses the
* Invocation span's resolved sampling metadata when available.
*
- *
Thread-safe: uses {@link ConcurrentHashMap} for span/scope storage since the SDK runs user code on multiple
- * threads.
+ *
Thread-safe within its invocation: the SDK runs user code on multiple threads, so the open-span registries are
+ * {@link ConcurrentHashMap}s. The invocation's identity needs no such protection — it is final state written before the
+ * SDK publishes the instance to those threads.
*/
-@ExclusivePluginGroup("durable-otel-view")
-public class ExecutionOtelPlugin implements DurableExecutionPlugin {
+@HandlerScoped(ExecutionOtelPlugin.HandlerScopeOpener.class)
+public final class ExecutionOtelPlugin implements DurableExecutionPlugin {
private static final Logger logger = LoggerFactory.getLogger(ExecutionOtelPlugin.class);
- private volatile SdkTracerProvider sdkTracerProvider;
- private volatile Tracer tracer;
+ // ─── Environment lifetime (shared with every other invocation's instance) ─────────────
+
private final DeterministicIdGenerator idGenerator;
- private final ContextExtractor contextExtractor;
private final boolean enableMdc;
private final String workflowSpanName;
- private final String instrumentationName;
-
- // Per-invocation state
- private volatile boolean tracingEnabled;
- private volatile Span invocationSpan;
- private volatile Span executionRootSpan;
- // The root handler opens and closes this context through the base invocation lifecycle.
- private Scope handlerScope;
- private volatile String durableExecutionArn;
-
- // Trace ID and flags of the execution trace, published together as one snapshot so readers never pair a trace ID
- // with mismatched flags.
- private volatile ExecutionTrace executionTrace;
- // The execution's single sampling intent for this invocation, computed once at onInvocationStart and attached to
- // every durable span's parent context so DurableSampler applies it (a resolved decision verbatim, or a deferral to
- // its own delegate) without re-invoking the configured sampler per span.
- private volatile DurableSamplingDecision.Intent samplingIntent;
-
- /** Immutable snapshot of the resolved execution trace, read atomically through a single volatile reference. */
- private record ExecutionTrace(String traceId, TraceFlags flags) {}
- // Between invocations the Workflow span exists only as a deterministic context that operations parent onto; the
- // recording span is started and ended in a single call on the terminal invocation, so it is never left open. The
- // execution ancestor and start time are retained so that span can be built at invocation end.
- private volatile SpanContext workflowSpanContext;
- private volatile SpanContext executionAncestor;
- private volatile Instant executionStartTime;
+ // ─── This invocation, all resolved in the constructor from its InvocationInfo ─────────
+
+ /** The provider used to flush before Lambda freezes; null when it is not visible to the application. */
+ private final SdkTracerProvider sdkTracerProvider;
+
+ /** Null when telemetry is disabled for this invocation, which makes every hook on this instance a no-op. */
+ private final Tracer tracer;
+
+ private final String durableExecutionArn;
+ private final Instant executionStartTime;
+
+ /** Trace ID and flags of the execution trace, resolved together so they can never be paired mismatched. */
+ private final ExecutionTrace executionTrace;
+
+ /**
+ * The execution's single sampling intent for this invocation, resolved once and attached to every durable span's
+ * parent context so DurableSampler applies it (a resolved decision verbatim, or a deferral to its own delegate)
+ * without re-invoking the configured sampler per span.
+ */
+ private final DurableSamplingDecision.Intent samplingIntent;
+
+ private final Span invocationSpan;
+ private final Span executionRootSpan;
+
+ /**
+ * The Workflow span exists as a deterministic context that operations parent onto; the recording span is started
+ * and ended in a single call on the terminal invocation, so it is never left open. The execution ancestor and start
+ * time are held so that span can be built at invocation end.
+ */
+ private final SpanContext workflowSpanContext;
+
+ private final SpanContext executionAncestor;
+
+ /**
+ * Set when this invocation ends; never cleared, because an instance is never reused. Read by the operation and user
+ * function hooks, which may run on other threads of this invocation, so that a straggler hook arriving after the
+ * spans have been ended does not open a new one — volatile for that publication.
+ */
+ private volatile boolean ended;
+
+ /** Immutable snapshot of the resolved execution trace. */
+ private record ExecutionTrace(String traceId, TraceFlags flags) {}
// Thread-safe storage for attempt spans/scopes (keyed by operationId + "-" + attempt)
private final ConcurrentHashMap attemptSpans = new ConcurrentHashMap<>();
@@ -141,91 +170,128 @@ private record ExecutionTrace(String traceId, TraceFlags flags) {}
private final ConcurrentHashMap operationStartTimes = new ConcurrentHashMap<>();
/**
- * Creates a Workflow-rooted OTel plugin with default settings: X-Ray context extraction, MDC enabled, root span
- * named {@code "Workflow"}.
+ * Returns a factory that creates one plugin instance per invocation against the ADOT Java agent's global provider,
+ * with default settings: X-Ray context extraction, MDC enabled, root span named {@code "Workflow"}.
*
- * Uses the provided tracer provider builder. For ADOT Java agent usage, prefer {@link #ExecutionOtelPlugin()}
- * with the plugin jar configured through {@code OTEL_JAVAAGENT_EXTENSIONS}.
+ *
{@code
+ * DurableConfig.builder().withPlugins(ExecutionOtelPlugin.factory()).build();
+ * }
+ *
+ * @return the per-invocation plugin factory to hand to {@code DurableConfig.Builder.withPlugins}
+ */
+ public static DurableExecutionPluginFactory factory() {
+ return factory(OtelPluginConfig.defaults());
+ }
+
+ /**
+ * Returns a factory that creates one plugin instance per invocation against the ADOT Java agent's global provider.
+ *
+ * The global provider is resolved when the first invocation's instance needs it. If the agent has not
+ * initialized it yet, that invocation emits no telemetry and the next invocation's instance resolves it again.
*
- * @param tracerProviderBuilder the tracer provider builder (its ID generator will be wrapped)
+ * @param config the plugin configuration
+ * @return the per-invocation plugin factory to hand to {@code DurableConfig.Builder.withPlugins}
*/
- public ExecutionOtelPlugin(SdkTracerProviderBuilder tracerProviderBuilder) {
- this(tracerProviderBuilder, OtelPluginConfig.defaults());
+ public static DurableExecutionPluginFactory factory(OtelPluginConfig config) {
+ var environment = OtelPluginEnvironment.forGlobalProvider(config);
+ return invocationFactory(environment);
}
/**
- * Creates a Workflow-rooted OTel plugin with default settings: X-Ray context extraction and MDC enabled.
+ * Returns a factory that creates one plugin instance per invocation against an application-owned tracer provider,
+ * with default settings: X-Ray context extraction, MDC enabled, root span named {@code "Workflow"}.
+ *
+ *
Customers configure exporters and span processors on the builder — the plugin handles ID generation. The
+ * provider is built once, here, and shared by every invocation's instance.
*
- *
Resolves {@code GlobalOpenTelemetry} at invocation start. If the ADOT Java agent has not initialized it yet,
- * telemetry is disabled for that invocation and resolution is retried on the next invocation.
+ * @param tracerProviderBuilder the tracer provider builder (its ID generator and sampler will be wrapped)
+ * @return the per-invocation plugin factory to hand to {@code DurableConfig.Builder.withPlugins}
*/
- public ExecutionOtelPlugin() {
- this(OtelPluginConfig.defaults());
+ public static DurableExecutionPluginFactory factory(SdkTracerProviderBuilder tracerProviderBuilder) {
+ return factory(tracerProviderBuilder, OtelPluginConfig.defaults());
}
/**
- * Creates a Workflow-rooted OTel plugin from the given tracer provider builder and configuration.
+ * Returns a factory that creates one plugin instance per invocation against an application-owned tracer provider.
*
*
Customers configure exporters and span processors on the builder; all other tunables (context extractor, MDC
- * toggle, Workflow span name, instrumentation scope name) come from {@link OtelPluginConfig}. Use
- * {@link OtelPluginConfig#builder()} for readable, named configuration:
+ * toggle, Workflow span name, instrumentation scope name) come from {@link OtelPluginConfig}:
*
*
{@code
- * var plugin = new ExecutionOtelPlugin(
+ * var factory = ExecutionOtelPlugin.factory(
* SdkTracerProvider.builder().addSpanProcessor(SimpleSpanProcessor.create(exporter)),
* OtelPluginConfig.builder().enableMdc(false).workflowSpanName("Workflow").build());
* }
*
- * @param tracerProviderBuilder the tracer provider builder (its ID generator will be wrapped)
+ * @param tracerProviderBuilder the tracer provider builder (its ID generator and sampler will be wrapped)
* @param config the plugin configuration
+ * @return the per-invocation plugin factory to hand to {@code DurableConfig.Builder.withPlugins}
*/
- public ExecutionOtelPlugin(SdkTracerProviderBuilder tracerProviderBuilder, OtelPluginConfig config) {
- OtelPluginSupport.requireSameThreadInvocationHooks();
- this.idGenerator = DeterministicIdGenerator.installOn(tracerProviderBuilder);
- // Wrap the configured sampler so durable spans use the execution's single precomputed decision.
- DurableSampler.installOn(tracerProviderBuilder);
-
- this.sdkTracerProvider = tracerProviderBuilder.build();
- this.tracer = sdkTracerProvider.get(config.instrumentationName());
- this.contextExtractor = config.contextExtractor();
- this.enableMdc = config.enableMdc();
- this.workflowSpanName = config.workflowSpanName();
- this.instrumentationName = config.instrumentationName();
+ public static DurableExecutionPluginFactory factory(
+ SdkTracerProviderBuilder tracerProviderBuilder, OtelPluginConfig config) {
+ var environment = OtelPluginEnvironment.forProviderBuilder(tracerProviderBuilder, config);
+ return invocationFactory(environment);
+ }
+
+ private static DurableExecutionPluginFactory invocationFactory(OtelPluginEnvironment environment) {
+ return new DurableExecutionPluginFactory() {
+ @Override
+ public String getExclusiveGroup() {
+ return "durable-otel-view";
+ }
+
+ @Override
+ public String toString() {
+ return "ExecutionOtelPlugin";
+ }
+
+ @Override
+ public DurableExecutionPlugin createPlugin(InvocationInfo info) {
+ return environment.createPlugin(() -> new ExecutionOtelPlugin(environment, info));
+ }
+ };
}
/**
- * Creates a Workflow-rooted OTel plugin from configuration alone (no caller-supplied tracer provider builder).
+ * Creates the instance that serves one invocation.
*
- * The config-only constructor uses the ADOT/global provider. Supply a {@code SdkTracerProviderBuilder} via the
- * two-arg constructor for an application-owned provider.
+ *
Everything this invocation's spans are keyed by is resolved here, from the {@code info} the factory received:
+ * the tracer binding, the extracted context, the canonical execution trace and its ancestor, the single sampling
+ * intent, the Invocation span, and the deferred Workflow span context. Resolving them in the constructor — before
+ * the SDK publishes this instance to the operation and user function threads — is what lets them be {@code final}
+ * rather than volatile per-invocation state.
*
- * @param config the plugin configuration
+ *
When the tracer cannot be bound, telemetry is disabled for this invocation: the span fields stay null and
+ * every hook returns immediately. The next invocation gets a new instance, which binds again.
*/
- public ExecutionOtelPlugin(OtelPluginConfig config) {
- OtelPluginSupport.requireSameThreadInvocationHooks();
- this.contextExtractor = config.contextExtractor();
+ private ExecutionOtelPlugin(OtelPluginEnvironment environment, InvocationInfo info) {
+ var config = environment.config();
+ this.idGenerator = environment.idGenerator();
this.enableMdc = config.enableMdc();
this.workflowSpanName = config.workflowSpanName();
- this.instrumentationName = config.instrumentationName();
- this.idGenerator = OtelPluginSupport.createDefaultIdGenerator();
- }
-
- // ─── Invocation hooks ────────────────────────────────────────────────
-
- @Override
- public void onInvocationStart(InvocationInfo info) {
- tracingEnabled = false;
- if (!bindTracer()) {
+ this.durableExecutionArn = info.durableExecutionArn();
+ this.executionStartTime = info.executionStartTime();
+
+ var setup = environment.bind("ExecutionOtelPlugin");
+ if (setup == null) {
+ this.sdkTracerProvider = null;
+ this.tracer = null;
+ this.samplingIntent = null;
+ this.executionTrace = null;
+ this.executionAncestor = null;
+ this.invocationSpan = null;
+ this.executionRootSpan = null;
+ this.workflowSpanContext = null;
return;
}
-
- this.durableExecutionArn = info.durableExecutionArn();
+ this.sdkTracerProvider = setup.sdkTracerProvider();
+ this.tracer = setup.tracer();
// Resolve the one execution ancestor both spans parent onto, so they share a stable-per-execution trace and a
// sampling decision.
- var extracted = contextExtractor.extract();
+ var extracted = config.contextExtractor().extract(info);
var canonicalTraceId =
- ExecutionTraceContext.canonicalTraceId(extracted, arn(), info.executionStartTime(), idGenerator);
+ ExecutionTraceContext.canonicalTraceId(extracted, arn(), executionStartTime, idGenerator);
// Resolve the execution's sampling decision once for this invocation as a full SamplingResult, then apply it to
// every durable span via DurableSampler. The execution ancestor's trace flags are derived from the same
// decision so a parent-based sampler stays consistent with it.
@@ -238,36 +304,31 @@ public void onInvocationStart(InvocationInfo info) {
Attributes.of(DURABLE_EXECUTION_ARN, arn()));
// An unresolved decision is supplied by the actual fallback root or the agent-side sampler. Do not invent
// a decision that bypasses an installed drop/rate-limit policy.
- samplingIntent = decision != null
+ this.samplingIntent = decision != null
? DurableSamplingDecision.Intent.resolved(decision)
: DurableSamplingDecision.Intent.deferred(canonicalTraceId);
var sampled = OtelPluginSupport.isSampled(decision);
var execCtx = ExecutionTraceContext.resolve(extracted, canonicalTraceId, arn(), idGenerator, () -> sampled);
- executionStartTime = info.executionStartTime();
+ // Materialize and immediately end the SDK-owned anchor before creating descendants or user code.
var root = DurableExecutionRoot.start(
tracer,
idGenerator,
execCtx.executionAncestor(),
- info.durableExecutionArn(),
+ durableExecutionArn,
executionStartTime,
samplingIntent,
OtelPluginSupport.usesLocalDurableSampler(sdkTracerProvider),
OtelPluginSupport.usesDurableSamplingBridge(sdkTracerProvider));
- try {
- DurableExecutionRoot.end(root, executionStartTime);
- } finally {
- // Publish only after end callbacks, so a reentrant End cannot flush from inside root.end().
- // Retain the ended root as a flush marker even if an onEnd processor throws during Start.
- executionRootSpan = root;
- }
+ DurableExecutionRoot.end(root, executionStartTime);
+ this.executionRootSpan = root;
// An unresolved plain fallback uses the real root's decision; explicit/ambient decisions and the
// legacy remote-parent path retain their existing behavior. Plain samplers still own each child decision.
if (executionRootSpan != null
&& (OtelPluginSupport.usesDurableSamplingBridge(sdkTracerProvider) || decision == null)) {
execCtx = new ExecutionTraceContext(executionRootSpan.getSpanContext());
}
- executionTrace = new ExecutionTrace(canonicalTraceId, execCtx.traceFlags());
- executionAncestor = execCtx.executionAncestor();
+ this.executionTrace = new ExecutionTrace(canonicalTraceId, execCtx.traceFlags());
+ this.executionAncestor = execCtx.executionAncestor();
// Invocation span — child of the ambient Lambda span when it is on the execution trace, otherwise a child of
// the execution ancestor so it stays within the same trace.
@@ -275,55 +336,65 @@ public void onInvocationStart(InvocationInfo info) {
var spanBuilder = tracer.spanBuilder("Invocation")
.setSpanKind(SpanKind.INTERNAL)
.setParent(invocationParent)
- .setAttribute(DURABLE_EXECUTION_ARN, info.durableExecutionArn())
+ .setAttribute(DURABLE_EXECUTION_ARN, durableExecutionArn)
.setAttribute(DURABLE_FIRST_INVOCATION, info.isFirstInvocation());
if (info.requestId() != null) {
spanBuilder.setAttribute(AttributeKey.stringKey("faas.invocation_id"), info.requestId());
}
- invocationSpan = startDurableSpan(spanBuilder);
+ this.invocationSpan = startDurableSpan(spanBuilder);
// Defer the recording Workflow span until terminal completion. The placeholder uses the Invocation span's
// resolved sampling metadata so operation parents/links match the span that is eventually exported.
- var workflowSpanId = idGenerator.generateWorkflowSpanId(info.durableExecutionArn());
var invocationContext = invocationSpan.getSpanContext();
- workflowSpanContext = SpanContext.create(
- canonicalTraceId, workflowSpanId, invocationContext.getTraceFlags(), invocationContext.getTraceState());
+ this.workflowSpanContext = SpanContext.create(
+ canonicalTraceId,
+ idGenerator.generateWorkflowSpanId(durableExecutionArn),
+ invocationContext.getTraceFlags(),
+ invocationContext.getTraceState());
+ }
- // Inject MDC on the handler thread so handler-level logs (between steps) have trace context.
- if (enableMdc) {
- MDC.put(
- MdcSpanEnricher.MDC_TRACE_ID,
- invocationSpan.getSpanContext().getTraceId());
+ // ─── Invocation hooks ────────────────────────────────────────────────
+
+ @Override
+ public void onInvocationStart(InvocationInfo info) {
+ // This invocation's identity, its Invocation span and its Workflow span context were resolved in the
+ // constructor, from the very InvocationInfo this hook receives. What is left is the MDC injection, which
+ // belongs
+ // here because it must run on the handler thread so handler-level logs between steps carry trace context.
+ if (invocationSpan == null || !enableMdc) {
+ return;
}
- tracingEnabled = true;
- handlerScope = activateHandlerContext();
+ MDC.put(MdcSpanEnricher.MDC_TRACE_ID, invocationSpan.getSpanContext().getTraceId());
}
- private Scope activateHandlerContext() {
+ /** JDK-only scope bridge; does not dispatch to coincidental subclass methods. */
+ public static final class HandlerScopeOpener implements Function {
+ @Override
+ public AutoCloseable apply(ExecutionOtelPlugin plugin) {
+ return plugin.activateHandlerContext();
+ }
+ }
+
+ private AutoCloseable activateHandlerContext() {
var trace = executionTrace;
- if (!tracingEnabled || trace == null) return null;
+ if (ended || tracer == null || trace == null) return null;
var ambient = Span.current().getSpanContext();
+ // Preserve a compatible ambient Lambda span. An absent or unrelated ambient span must not leave
+ // handler instrumentation outside the durable execution's canonical trace.
if (ambient.isValid() && trace.traceId().equals(ambient.getTraceId())) return Scope.noop();
return Span.wrap(workflowSpanContext).makeCurrent();
}
@Override
public void onInvocationEnd(InvocationEndInfo info) {
- var scope = handlerScope;
- handlerScope = null;
- OtelPluginSupport.runInvocationEnd(scope, () -> endInvocation(info));
- }
-
- private void endInvocation(InvocationEndInfo info) {
- var root = executionRootSpan;
- var provider = sdkTracerProvider;
- var shouldFlush = tracingEnabled || root != null;
- executionRootSpan = null; // The anchor already ended at Start; flush this invocation only once.
+ if (ended) return;
+ ended = true;
+ var shouldFlush = invocationSpan != null || executionRootSpan != null;
OtelPluginSupport.finishWithRootCleanup(() -> finishInvocationEnd(info), () -> {
- if (shouldFlush && provider != null) {
- var flushResult = provider.forceFlush().join(5, TimeUnit.SECONDS);
+ if (shouldFlush && sdkTracerProvider != null) {
+ var flushResult = sdkTracerProvider.forceFlush().join(5, TimeUnit.SECONDS);
if (!flushResult.isSuccess())
logger.warn("OTel span flush failed or timed out — some spans may be lost");
}
@@ -331,39 +402,28 @@ private void endInvocation(InvocationEndInfo info) {
}
private void finishInvocationEnd(InvocationEndInfo info) {
- if (!tracingEnabled) {
- return;
- }
- tracingEnabled = false;
-
+ if (invocationSpan == null) return;
// Clear invocation-level MDC
if (enableMdc) {
MdcSpanEnricher.clear();
}
- // Drop placeholder state. Open operations have no recording span to abandon.
- operationContexts.clear();
- operationStartTimes.clear();
-
- // End open recording spans now, but leave a running worker's thread-bound scope for its own user-function End.
- // Only scopes owned by this finalization thread can be closed here.
+ // Close scopes owned by this thread, retaining other owners' scopes until their matching end hooks run.
attemptScopes.closeCurrentThread();
for (var span : attemptSpans.values()) {
span.end();
}
- attemptSpans.clear();
+ // Every attempt span has ended. Only unfinished owners may retain this invocation's instance for scope
+ // cleanup; these registries are never recycled for another invocation.
// End the invocation span every invocation.
- if (invocationSpan != null) {
- invocationSpan.setAttribute(
- DURABLE_INVOCATION_STATUS, info.invocationStatus().name());
- applyInvocationStatus(invocationSpan, info);
- invocationSpan.end();
- invocationSpan = null;
- }
+ invocationSpan.setAttribute(
+ DURABLE_INVOCATION_STATUS, info.invocationStatus().name());
+ applyInvocationStatus(invocationSpan, info);
+ invocationSpan.end();
// Materialize the Workflow span only on terminal status.
- if (isTerminal(info) && workflowSpanContext != null && executionAncestor != null) {
+ if (isTerminal(info)) {
var workflowSpanBuilder = tracer.spanBuilder(workflowSpanName)
.setSpanKind(SpanKind.INTERNAL)
.setParent(withDurableDecision(Context.root().with(Span.wrap(executionAncestor))))
@@ -387,17 +447,13 @@ private void finishInvocationEnd(InvocationEndInfo info) {
}
workflowSpan.end();
}
- workflowSpanContext = null;
- executionAncestor = null;
- executionStartTime = null;
- samplingIntent = null;
}
/** Produces metadata without creating spans or changing the configured provider/resource. */
@Override
public String providePropagationMetadata(
String executionArn, String operationId, String parentOperationId, String targetFunctionName) {
- if (!tracingEnabled || !executionArn.equals(durableExecutionArn)) return null;
+ if (disabled() || !executionArn.equals(durableExecutionArn)) return null;
var trace = executionTrace;
if (trace == null) return null;
// These contexts are deferred placeholders, not sampled operation spans. A replacement sampler may decide
@@ -419,7 +475,7 @@ public String providePropagationMetadata(
@Override
public void onOperationStart(OperationInfo info) {
- if (!tracingEnabled) return;
+ if (disabled()) return;
if (info.id() == null) return;
// Retain only a deterministic placeholder. Its flags/state come from the Invocation span's resolved sampling
@@ -437,7 +493,7 @@ public void onOperationStart(OperationInfo info) {
@Override
public void onOperationEnd(OperationEndInfo info) {
- if (!tracingEnabled) return;
+ if (disabled()) return;
if (info.id() == null) return;
// Start and end the operation's single span here, using its deterministic span ID and linking to the
@@ -490,7 +546,7 @@ public void onOperationEnd(OperationEndInfo info) {
@Override
public void onUserFunctionStart(UserFunctionStartInfo info) {
- if (!tracingEnabled) return;
+ if (disabled()) return;
// Skip attempt spans for CONTEXT operations — they are a scoping construct, not a retriable unit of work. Still
// make the operation's context current so auto-instrumented calls become children of the (deferred) operation
@@ -498,7 +554,8 @@ public void onUserFunctionStart(UserFunctionStartInfo info) {
if ("CONTEXT".equals(info.type())) {
var operationContext = operationContexts.get(info.id());
if (operationContext != null) {
- var scope = Span.wrap(operationContext).makeCurrent();
+ var scope = OtelPluginSupport.onOwnerThread(
+ Span.wrap(operationContext).makeCurrent());
var key = attemptKey(info.id(), info.attempt());
attemptScopes.put(key, scope);
}
@@ -538,7 +595,7 @@ public void onUserFunctionStart(UserFunctionStartInfo info) {
attemptSpans.put(key, span);
// Make span current on this thread so auto-instrumented calls become children
- var scope = span.makeCurrent();
+ var scope = OtelPluginSupport.onOwnerThread(span.makeCurrent());
attemptScopes.put(key, scope);
if (enableMdc) {
@@ -548,10 +605,14 @@ public void onUserFunctionStart(UserFunctionStartInfo info) {
@Override
public void onUserFunctionEnd(UserFunctionEndInfo info) {
+ if (invocationSpan == null) return;
+
var key = attemptKey(info.id(), info.attempt());
- // A running user function can finish after invocation End; its scope still belongs to this thread.
+
+ // A late user-function End still closes the scope on the owning thread.
attemptScopes.close(key);
- if (!tracingEnabled) return;
+
+ if (ended) return;
// CONTEXT operations don't have attempt spans — scope cleanup is all we need
if ("CONTEXT".equals(info.type())) {
@@ -581,22 +642,12 @@ public void onUserFunctionEnd(UserFunctionEndInfo info) {
// ─── Helpers ─────────────────────────────────────────────────────────
- private boolean bindTracer() {
- if (tracer != null) {
- return true;
- }
- synchronized (this) {
- if (tracer != null) {
- return true;
- }
- var setup = OtelPluginSupport.tryResolveGlobalProvider(instrumentationName, "ExecutionOtelPlugin");
- if (setup == null) {
- return false;
- }
- sdkTracerProvider = setup.sdkTracerProvider();
- tracer = setup.tracer();
- return true;
- }
+ /**
+ * True when this instance emits no telemetry: either the tracer could not be bound for this invocation, or the
+ * invocation has already ended and its spans are closed.
+ */
+ private boolean disabled() {
+ return invocationSpan == null || ended;
}
private void applyInvocationStatus(Span span, InvocationEndInfo info) {
@@ -695,17 +746,11 @@ private Context withDurableDecision(Context context) {
}
private TraceFlags effectiveTraceFlags() {
- var invocation = invocationSpan;
- if (invocation != null) {
- return invocation.getSpanContext().getTraceFlags();
- }
- var trace = executionTrace;
- return trace != null ? trace.flags() : TraceFlags.getDefault();
+ return invocationSpan.getSpanContext().getTraceFlags();
}
private TraceState effectiveTraceState() {
- var invocation = invocationSpan;
- return invocation != null ? invocation.getSpanContext().getTraceState() : TraceState.getDefault();
+ return invocationSpan.getSpanContext().getTraceState();
}
/**
diff --git a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/ExecutionOtelPluginProvider.java b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/ExecutionOtelPluginProvider.java
index 011e1dc028..01aee63c3c 100644
--- a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/ExecutionOtelPluginProvider.java
+++ b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/ExecutionOtelPluginProvider.java
@@ -3,30 +3,32 @@
package software.amazon.lambda.durable.otel;
import software.amazon.lambda.durable.plugin.DurableExecutionPlugin;
+import software.amazon.lambda.durable.plugin.DurableExecutionPluginFactory;
import software.amazon.lambda.durable.plugin.DurableExecutionPluginProvider;
+import software.amazon.lambda.durable.plugin.InvocationInfo;
/**
* Dynamically loads {@link ExecutionOtelPlugin} when {@code DURABLE_EXECUTION_PLUGINS} contains {@code otel-execution}.
+ *
+ * The provider is itself the per-invocation factory: it holds the environment-lifetime state (the ADOT global
+ * provider binding, the ID generator) once and creates one plugin instance per invocation from it.
*/
public final class ExecutionOtelPluginProvider implements DurableExecutionPluginProvider {
+ private final DurableExecutionPluginFactory factory = ExecutionOtelPlugin.factory();
+
@Override
public String getName() {
return "otel-execution";
}
@Override
- public int getApiVersion() {
- return API_VERSION;
- }
-
- @Override
- public Class extends DurableExecutionPlugin> getPluginType() {
- return ExecutionOtelPlugin.class;
+ public DurableExecutionPlugin createPlugin(InvocationInfo invocationInfo) {
+ return factory.createPlugin(invocationInfo);
}
@Override
- public DurableExecutionPlugin createPlugin() {
- return new ExecutionOtelPlugin();
+ public String getExclusiveGroup() {
+ return factory.getExclusiveGroup();
}
}
diff --git a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/InvocationOtelPlugin.java b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/InvocationOtelPlugin.java
index a821787978..c1ecbc4e6d 100644
--- a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/InvocationOtelPlugin.java
+++ b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/InvocationOtelPlugin.java
@@ -22,11 +22,13 @@
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import software.amazon.lambda.durable.plugin.DurableExecutionPlugin;
-import software.amazon.lambda.durable.plugin.ExclusivePluginGroup;
+import software.amazon.lambda.durable.plugin.DurableExecutionPluginFactory;
+import software.amazon.lambda.durable.plugin.HandlerScoped;
import software.amazon.lambda.durable.plugin.InvocationEndInfo;
import software.amazon.lambda.durable.plugin.InvocationInfo;
import software.amazon.lambda.durable.plugin.OperationEndInfo;
@@ -62,9 +64,18 @@
*
Tracing: Active (to populate {@code _X_AMZN_TRACE_ID})
*
*
- * When using {@link #InvocationOtelPlugin()}, the plugin resolves the global provider at invocation start. If the
- * OpenTelemetry Java agent is not initialized yet, telemetry is disabled for that entire invocation and provider
- * resolution is retried on the next invocation.
+ *
Lifetime. One instance serves exactly one Lambda invocation: {@link #factory()} and its overloads return a
+ * {@link DurableExecutionPluginFactory} that the SDK calls once per invocation. Invocation-level references are
+ * released at return; an unfinished operation retains its matching cleanup until its owner unwinds. Everything about
+ * the invocation — the execution ARN, the resolved execution trace and ancestor, the sampling intent, the Invocation
+ * span, the deferred Workflow span context — is therefore a {@code final} field, resolved in the constructor from the
+ * {@link InvocationInfo} the factory receives (the same instance {@link #onInvocationStart(InvocationInfo)} then
+ * receives). Nothing is reset between invocations because nothing is carried between them.
+ *
+ *
What belongs to the execution environment stays in the factory's {@link OtelPluginEnvironment}: the configuration,
+ * the ID generator, and either the application-owned tracer provider (built once) or the lazily resolved ADOT global
+ * provider. On the agent path, an invocation whose instance cannot resolve the global provider emits no telemetry at
+ * all, and the next invocation's instance resolves it again.
*
*
X-Ray console limitation: In the X-Ray "Segments Timeline" ungrouped view, the plugin's spans (Invocation,
* operation, attempt) do not appear as nested subsegments of the Lambda platform segment. This is a known limitation of
@@ -73,43 +84,58 @@
* view to inspect parent-child relationships within the shared execution trace and the links between operation spans
* and the Workflow span.
*
- *
Thread-safe: uses {@link ConcurrentHashMap} for span/scope storage since the SDK runs user code on multiple
- * threads.
+ *
Thread-safe within its invocation: the SDK runs user code on multiple threads, so the open-span registries are
+ * {@link ConcurrentHashMap}s. The invocation's identity needs no such protection — it is final state written before the
+ * SDK publishes the instance to those threads.
*/
-@ExclusivePluginGroup("durable-otel-view")
-public class InvocationOtelPlugin implements DurableExecutionPlugin {
+@HandlerScoped(InvocationOtelPlugin.HandlerScopeOpener.class)
+public final class InvocationOtelPlugin implements DurableExecutionPlugin {
private static final Logger logger = LoggerFactory.getLogger(InvocationOtelPlugin.class);
- private volatile SdkTracerProvider sdkTracerProvider;
- private volatile Tracer tracer;
+ // ─── Environment lifetime (shared with every other invocation's instance) ─────────────
+
private final DeterministicIdGenerator idGenerator;
- private final ContextExtractor contextExtractor;
private final boolean enableMdc;
private final String workflowSpanName;
- private final String instrumentationName;
-
- // Per-invocation state
- private volatile boolean tracingEnabled;
- private volatile Span invocationSpan;
- private volatile Span executionRootSpan;
- // The root handler opens and closes this context through the base invocation lifecycle.
- private Scope handlerScope;
- private volatile String durableExecutionArn;
- // Trace ID and flags of the execution trace, published together as one snapshot so readers never pair a trace ID
- // with mismatched flags.
- private volatile ExecutionTrace executionTrace;
- // The execution's single sampling intent for this invocation, computed once at onInvocationStart and attached to
- // every durable span's parent context so DurableSampler applies it (a resolved decision verbatim, or a deferral to
- // its own delegate) without re-invoking the configured sampler per span.
- private volatile DurableSamplingDecision.Intent samplingIntent;
-
- // Deferred Workflow placeholder; the recording span is emitted only on terminal invocation.
- private volatile SpanContext workflowSpanContext;
- private volatile SpanContext executionAncestor;
- private volatile Instant executionStartTime;
-
- /** Immutable snapshot of the resolved execution trace, read atomically through a single volatile reference. */
+
+ // ─── This invocation, all resolved in the constructor from its InvocationInfo ─────────
+
+ /** The provider used to flush before Lambda freezes; null when it is not visible to the application. */
+ private final SdkTracerProvider sdkTracerProvider;
+
+ /** Null when telemetry is disabled for this invocation, which makes every hook on this instance a no-op. */
+ private final Tracer tracer;
+
+ private final String durableExecutionArn;
+ private final Instant executionStartTime;
+
+ /** Trace ID and flags of the execution trace, resolved together so they can never be paired mismatched. */
+ private final ExecutionTrace executionTrace;
+
+ /**
+ * The execution's single sampling intent for this invocation, resolved once and attached to every durable span's
+ * parent context so DurableSampler applies it (a resolved decision verbatim, or a deferral to its own delegate)
+ * without re-invoking the configured sampler per span.
+ */
+ private final DurableSamplingDecision.Intent samplingIntent;
+
+ private final SpanContext executionAncestor;
+
+ private final Span invocationSpan;
+ private final Span executionRootSpan;
+
+ /** Deferred Workflow placeholder; the recording span is emitted only on terminal invocation. */
+ private final SpanContext workflowSpanContext;
+
+ /**
+ * Set when this invocation ends; never cleared, because an instance is never reused. Read by the operation and user
+ * function hooks, which may run on other threads of this invocation, so that a straggler hook arriving after the
+ * spans have been ended does not open a new one — volatile for that publication.
+ */
+ private volatile boolean ended;
+
+ /** Immutable snapshot of the resolved execution trace. */
private record ExecutionTrace(String traceId, TraceFlags flags) {}
// Thread-safe storage for operation spans (keyed by operationId) — open spans that need ending
@@ -126,99 +152,134 @@ private record ExecutionTrace(String traceId, TraceFlags flags) {}
private final ConcurrentLinkedDeque operationStartOrder = new ConcurrentLinkedDeque<>();
/**
- * Creates an OTel plugin with default settings: X-Ray context extraction, MDC enabled.
- *
- * Uses the provided tracer provider builder. Customers configure exporters and span processors on the builder —
- * the plugin handles ID generation.
- *
- *
For ADOT Java agent usage, prefer {@link #InvocationOtelPlugin()} with the plugin jar configured through
- * {@code OTEL_JAVAAGENT_EXTENSIONS}. Use this builder constructor when you want to own the exporter pipeline:
+ * Returns a factory that creates one plugin instance per invocation against the ADOT Java agent's global provider,
+ * with default settings: X-Ray context extraction and MDC enabled.
*
*
{@code
- * var exporter = LoggingSpanExporter.create();
- * var plugin = new InvocationOtelPlugin(
- * SdkTracerProvider.builder().addSpanProcessor(SimpleSpanProcessor.create(exporter)));
+ * DurableConfig.builder().withPlugins(InvocationOtelPlugin.factory()).build();
* }
*
- * @param tracerProviderBuilder the tracer provider builder (its ID generator will be wrapped)
+ * @return the per-invocation plugin factory to hand to {@code DurableConfig.Builder.withPlugins}
+ */
+ public static DurableExecutionPluginFactory factory() {
+ return factory(OtelPluginConfig.defaults());
+ }
+
+ /**
+ * Returns a factory that creates one plugin instance per invocation against the ADOT Java agent's global provider.
+ *
+ * The global provider is resolved when the first invocation's instance needs it. If the agent has not
+ * initialized it yet, that invocation emits no telemetry and the next invocation's instance resolves it again.
+ *
+ * @param config the plugin configuration
+ * @return the per-invocation plugin factory to hand to {@code DurableConfig.Builder.withPlugins}
*/
- public InvocationOtelPlugin(SdkTracerProviderBuilder tracerProviderBuilder) {
- this(tracerProviderBuilder, OtelPluginConfig.defaults());
+ public static DurableExecutionPluginFactory factory(OtelPluginConfig config) {
+ var environment = OtelPluginEnvironment.forGlobalProvider(config);
+ return invocationFactory(environment);
}
/**
- * Creates an OTel plugin with default settings: X-Ray context extraction and MDC enabled.
+ * Returns a factory that creates one plugin instance per invocation against an application-owned tracer provider,
+ * with default settings: X-Ray context extraction and MDC enabled.
+ *
+ *
Customers configure exporters and span processors on the builder — the plugin handles ID generation. The
+ * provider is built once, here, and shared by every invocation's instance:
+ *
+ *
{@code
+ * var exporter = LoggingSpanExporter.create();
+ * var factory = InvocationOtelPlugin.factory(
+ * SdkTracerProvider.builder().addSpanProcessor(SimpleSpanProcessor.create(exporter)));
+ * }
*
- * Resolves {@code GlobalOpenTelemetry} at invocation start. If the ADOT Java agent has not initialized it yet,
- * telemetry is disabled for that invocation and resolution is retried on the next invocation.
+ * @param tracerProviderBuilder the tracer provider builder (its ID generator and sampler will be wrapped)
+ * @return the per-invocation plugin factory to hand to {@code DurableConfig.Builder.withPlugins}
*/
- public InvocationOtelPlugin() {
- this(OtelPluginConfig.defaults());
+ public static DurableExecutionPluginFactory factory(SdkTracerProviderBuilder tracerProviderBuilder) {
+ return factory(tracerProviderBuilder, OtelPluginConfig.defaults());
}
/**
- * Creates an OTel plugin from the given tracer provider builder and configuration.
+ * Returns a factory that creates one plugin instance per invocation against an application-owned tracer provider.
*
*
Customers configure exporters and span processors on the builder; all other tunables (context extractor, MDC
- * toggle, Workflow span name, instrumentation scope name) come from {@link OtelPluginConfig}. Use
- * {@link OtelPluginConfig#builder()} for readable, named configuration:
+ * toggle, Workflow span name, instrumentation scope name) come from {@link OtelPluginConfig}:
*
*
{@code
- * var plugin = new InvocationOtelPlugin(
+ * var factory = InvocationOtelPlugin.factory(
* SdkTracerProvider.builder().addSpanProcessor(SimpleSpanProcessor.create(exporter)),
* OtelPluginConfig.builder().enableMdc(false).workflowSpanName("Workflow").build());
* }
*
- * @param tracerProviderBuilder the tracer provider builder (its ID generator will be wrapped)
+ * @param tracerProviderBuilder the tracer provider builder (its ID generator and sampler will be wrapped)
* @param config the plugin configuration
+ * @return the per-invocation plugin factory to hand to {@code DurableConfig.Builder.withPlugins}
*/
- public InvocationOtelPlugin(SdkTracerProviderBuilder tracerProviderBuilder, OtelPluginConfig config) {
- OtelPluginSupport.requireSameThreadInvocationHooks();
- this.idGenerator = DeterministicIdGenerator.installOn(tracerProviderBuilder);
- // Wrap the configured sampler so durable spans use the execution's single precomputed decision.
- DurableSampler.installOn(tracerProviderBuilder);
-
- this.sdkTracerProvider = tracerProviderBuilder.build();
- this.tracer = sdkTracerProvider.get(config.instrumentationName());
- this.contextExtractor = config.contextExtractor();
- this.enableMdc = config.enableMdc();
- this.workflowSpanName = config.workflowSpanName();
- this.instrumentationName = config.instrumentationName();
+ public static DurableExecutionPluginFactory factory(
+ SdkTracerProviderBuilder tracerProviderBuilder, OtelPluginConfig config) {
+ var environment = OtelPluginEnvironment.forProviderBuilder(tracerProviderBuilder, config);
+ return invocationFactory(environment);
+ }
+
+ private static DurableExecutionPluginFactory invocationFactory(OtelPluginEnvironment environment) {
+ return new DurableExecutionPluginFactory() {
+ @Override
+ public String getExclusiveGroup() {
+ return "durable-otel-view";
+ }
+
+ @Override
+ public String toString() {
+ return "InvocationOtelPlugin";
+ }
+
+ @Override
+ public DurableExecutionPlugin createPlugin(InvocationInfo info) {
+ return environment.createPlugin(() -> new InvocationOtelPlugin(environment, info));
+ }
+ };
}
/**
- * Creates an OTel plugin from configuration alone (no caller-supplied tracer provider builder).
+ * Creates the instance that serves one invocation.
*
- * The config-only constructor uses the ADOT/global provider. Supply a {@code SdkTracerProviderBuilder} via the
- * two-arg constructor for an application-owned provider.
+ *
Everything this invocation's spans are keyed by is resolved here, from the {@code info} the factory received:
+ * the tracer binding, the extracted context, the canonical execution trace and its ancestor, the single sampling
+ * intent, the Invocation span, and the deferred Workflow span context. Resolving them in the constructor — before
+ * the SDK publishes this instance to the operation and user function threads — is what lets them be {@code final}
+ * rather than volatile per-invocation state.
*
- * @param config the plugin configuration
+ *
When the tracer cannot be bound, telemetry is disabled for this invocation: the span fields stay null and
+ * every hook returns immediately. The next invocation gets a new instance, which binds again.
*/
- public InvocationOtelPlugin(OtelPluginConfig config) {
- OtelPluginSupport.requireSameThreadInvocationHooks();
- this.contextExtractor = config.contextExtractor();
+ private InvocationOtelPlugin(OtelPluginEnvironment environment, InvocationInfo info) {
+ var config = environment.config();
+ this.idGenerator = environment.idGenerator();
this.enableMdc = config.enableMdc();
this.workflowSpanName = config.workflowSpanName();
- this.instrumentationName = config.instrumentationName();
- this.idGenerator = OtelPluginSupport.createDefaultIdGenerator();
- }
-
- // ─── Invocation hooks ────────────────────────────────────────────────
-
- @Override
- public void onInvocationStart(InvocationInfo info) {
- tracingEnabled = false;
- if (!bindTracer()) {
+ this.durableExecutionArn = info.durableExecutionArn();
+ this.executionStartTime = info.executionStartTime();
+
+ var setup = environment.bind("InvocationOtelPlugin");
+ if (setup == null) {
+ this.sdkTracerProvider = null;
+ this.tracer = null;
+ this.samplingIntent = null;
+ this.executionTrace = null;
+ this.executionAncestor = null;
+ this.invocationSpan = null;
+ this.executionRootSpan = null;
+ this.workflowSpanContext = null;
return;
}
+ this.sdkTracerProvider = setup.sdkTracerProvider();
+ this.tracer = setup.tracer();
- this.durableExecutionArn = info.durableExecutionArn();
-
- var extracted = contextExtractor.extract();
+ var extracted = config.contextExtractor().extract(info);
// Resolve the execution ancestor the Workflow span parents onto so it joins the stable-per-execution trace.
- var canonicalTraceId = ExecutionTraceContext.canonicalTraceId(
- extracted, info.durableExecutionArn(), info.executionStartTime(), idGenerator);
+ var canonicalTraceId =
+ ExecutionTraceContext.canonicalTraceId(extracted, durableExecutionArn, executionStartTime, idGenerator);
// Resolve the execution's sampling decision once for this invocation as a full SamplingResult, then apply it to
// every durable span via DurableSampler (see below). The execution ancestor's trace flags are derived from the
// same decision so a parent-based sampler stays consistent with it.
@@ -231,37 +292,32 @@ public void onInvocationStart(InvocationInfo info) {
Attributes.of(DURABLE_EXECUTION_ARN, info.durableExecutionArn()));
// An unresolved decision is supplied by the actual fallback root or the agent-side sampler. Do not invent
// a decision that bypasses an installed drop/rate-limit policy.
- samplingIntent = decision != null
+ this.samplingIntent = decision != null
? DurableSamplingDecision.Intent.resolved(decision)
: DurableSamplingDecision.Intent.deferred(canonicalTraceId);
var sampled = OtelPluginSupport.isSampled(decision);
var execCtx = ExecutionTraceContext.resolve(
- extracted, canonicalTraceId, info.durableExecutionArn(), idGenerator, () -> sampled);
- executionStartTime = info.executionStartTime();
+ extracted, canonicalTraceId, durableExecutionArn, idGenerator, () -> sampled);
+ // Materialize and immediately end the SDK-owned anchor before creating descendants or user code.
var root = DurableExecutionRoot.start(
tracer,
idGenerator,
execCtx.executionAncestor(),
- info.durableExecutionArn(),
+ durableExecutionArn,
executionStartTime,
samplingIntent,
OtelPluginSupport.usesLocalDurableSampler(sdkTracerProvider),
OtelPluginSupport.usesDurableSamplingBridge(sdkTracerProvider));
- try {
- DurableExecutionRoot.end(root, executionStartTime);
- } finally {
- // Publish only after end callbacks, so a reentrant End cannot flush from inside root.end().
- // Retain the ended root as a flush marker even if an onEnd processor throws during Start.
- executionRootSpan = root;
- }
+ DurableExecutionRoot.end(root, executionStartTime);
+ this.executionRootSpan = root;
// An unresolved plain fallback uses the real root's decision; explicit/ambient decisions and the
// legacy remote-parent path retain their existing behavior. Plain samplers still own each child decision.
if (executionRootSpan != null
&& (OtelPluginSupport.usesDurableSamplingBridge(sdkTracerProvider) || decision == null)) {
execCtx = new ExecutionTraceContext(executionRootSpan.getSpanContext());
}
- executionTrace = new ExecutionTrace(canonicalTraceId, execCtx.traceFlags());
- executionAncestor = execCtx.executionAncestor();
+ this.executionTrace = new ExecutionTrace(canonicalTraceId, execCtx.traceFlags());
+ this.executionAncestor = execCtx.executionAncestor();
// Invocation span parent — the same-trace ambient span when available, then the execution ancestor, so the
// Invocation span stays on the execution trace.
@@ -271,56 +327,65 @@ public void onInvocationStart(InvocationInfo info) {
var spanBuilder = tracer.spanBuilder("Invocation")
.setSpanKind(SpanKind.INTERNAL)
.setParent(parentContext)
- .setAttribute(DURABLE_EXECUTION_ARN, info.durableExecutionArn())
+ .setAttribute(DURABLE_EXECUTION_ARN, durableExecutionArn)
.setAttribute(DURABLE_FIRST_INVOCATION, info.isFirstInvocation());
if (info.requestId() != null) {
spanBuilder.setAttribute(AttributeKey.stringKey("faas.invocation_id"), info.requestId());
}
- invocationSpan = startDurableSpan(spanBuilder);
+ this.invocationSpan = startDurableSpan(spanBuilder);
// Defer the recording Workflow span until terminal completion. The placeholder uses the Invocation span's
// resolved sampling metadata so operation links match the span that is eventually exported.
- var workflowSpanId = idGenerator.generateWorkflowSpanId(info.durableExecutionArn());
var invocationContext = invocationSpan.getSpanContext();
- workflowSpanContext = SpanContext.create(
- canonicalTraceId, workflowSpanId, invocationContext.getTraceFlags(), invocationContext.getTraceState());
+ this.workflowSpanContext = SpanContext.create(
+ canonicalTraceId,
+ idGenerator.generateWorkflowSpanId(durableExecutionArn),
+ invocationContext.getTraceFlags(),
+ invocationContext.getTraceState());
+ }
- // Inject MDC on the handler thread so handler-level logs (between steps) have trace context.
- // This runs on the same thread as context.getLogger() calls in the handler.
- if (enableMdc) {
- MDC.put(
- MdcSpanEnricher.MDC_TRACE_ID,
- invocationSpan.getSpanContext().getTraceId());
+ // ─── Invocation hooks ────────────────────────────────────────────────
+
+ @Override
+ public void onInvocationStart(InvocationInfo info) {
+ // This invocation's identity and its Invocation span were resolved in the constructor, from the very
+ // InvocationInfo this hook receives. What is left is the MDC injection, which belongs here because it must run
+ // on the handler thread — the same thread as the context.getLogger() calls in the handler — so handler-level
+ // logs between steps carry trace context.
+ if (invocationSpan == null || !enableMdc) {
+ return;
}
- tracingEnabled = true;
- handlerScope = activateHandlerContext();
+ MDC.put(MdcSpanEnricher.MDC_TRACE_ID, invocationSpan.getSpanContext().getTraceId());
}
- private Scope activateHandlerContext() {
+ /** JDK-only scope bridge; does not dispatch to coincidental subclass methods. */
+ public static final class HandlerScopeOpener implements Function {
+ @Override
+ public AutoCloseable apply(InvocationOtelPlugin plugin) {
+ return plugin.activateHandlerContext();
+ }
+ }
+
+ private AutoCloseable activateHandlerContext() {
var trace = executionTrace;
- if (!tracingEnabled || trace == null) return null;
+ if (ended || tracer == null || trace == null) return null;
var ambient = Span.current().getSpanContext();
+ // Preserve a compatible ambient Lambda span. An absent or unrelated ambient span must not leave
+ // handler instrumentation outside the durable execution's canonical trace.
if (ambient.isValid() && trace.traceId().equals(ambient.getTraceId())) return Scope.noop();
return invocationSpan.makeCurrent();
}
@Override
public void onInvocationEnd(InvocationEndInfo info) {
- var scope = handlerScope;
- handlerScope = null;
- OtelPluginSupport.runInvocationEnd(scope, () -> endInvocation(info));
- }
-
- private void endInvocation(InvocationEndInfo info) {
- var root = executionRootSpan;
- var provider = sdkTracerProvider;
- var shouldFlush = tracingEnabled || root != null;
- executionRootSpan = null; // The anchor already ended at Start; flush this invocation only once.
+ if (ended) return;
+ ended = true;
+ var shouldFlush = invocationSpan != null || executionRootSpan != null;
OtelPluginSupport.finishWithRootCleanup(() -> finishInvocationEnd(info), () -> {
- if (shouldFlush && provider != null) {
- var flushResult = provider.forceFlush().join(5, TimeUnit.SECONDS);
+ if (shouldFlush && sdkTracerProvider != null) {
+ var flushResult = sdkTracerProvider.forceFlush().join(5, TimeUnit.SECONDS);
if (!flushResult.isSuccess())
logger.warn("OTel span flush failed or timed out — some spans may be lost");
}
@@ -328,18 +393,12 @@ private void endInvocation(InvocationEndInfo info) {
}
private void finishInvocationEnd(InvocationEndInfo info) {
- if (!tracingEnabled) {
- return;
- }
- tracingEnabled = false;
-
+ if (invocationSpan == null) return;
// Clear invocation-level MDC (set in onInvocationStart on the handler thread)
if (enableMdc) {
MdcSpanEnricher.clear();
}
- if (invocationSpan == null) return;
-
endOpenSpansChildFirst();
// End invocation span
@@ -369,10 +428,9 @@ private void finishInvocationEnd(InvocationEndInfo info) {
}
invocationSpan.end();
- invocationSpan = null;
// Materialize the Workflow span only on terminal status.
- if (isTerminal(info) && workflowSpanContext != null && executionAncestor != null) {
+ if (isTerminal(info)) {
var workflowSpanBuilder = tracer.spanBuilder(workflowSpanName)
.setSpanKind(SpanKind.INTERNAL)
.setParent(withDurableDecision(Context.root().with(Span.wrap(executionAncestor))))
@@ -396,17 +454,13 @@ private void finishInvocationEnd(InvocationEndInfo info) {
}
workflowSpan.end();
}
- workflowSpanContext = null;
- executionAncestor = null;
- executionStartTime = null;
- samplingIntent = null;
}
/** Produces metadata without creating spans or changing the configured provider/resource. */
@Override
public String providePropagationMetadata(
String executionArn, String operationId, String parentOperationId, String targetFunctionName) {
- if (!tracingEnabled || !executionArn.equals(durableExecutionArn)) return null;
+ if (disabled() || !executionArn.equals(durableExecutionArn)) return null;
var trace = executionTrace;
if (trace == null) return null;
// An observed operation's actual context is authoritative, including invocation-view continuation segments.
@@ -430,7 +484,7 @@ public String providePropagationMetadata(
@Override
public void onOperationStart(OperationInfo info) {
- if (!tracingEnabled) return;
+ if (disabled()) return;
if (info.id() == null) return;
var parentContext = resolveParentContext(info.parentId());
@@ -471,7 +525,7 @@ public void onOperationStart(OperationInfo info) {
@Override
public void onOperationEnd(OperationEndInfo info) {
- if (!tracingEnabled) return;
+ if (disabled()) return;
if (info.id() == null) return;
var span = operationSpans.remove(info.id());
@@ -526,7 +580,7 @@ public void onOperationEnd(OperationEndInfo info) {
@Override
public void onUserFunctionStart(UserFunctionStartInfo info) {
- if (!tracingEnabled) return;
+ if (disabled()) return;
// Skip attempt spans for CONTEXT operations — they are a scoping construct, not a
// retriable unit of work, so attempt number/outcome attributes don't apply.
@@ -535,7 +589,7 @@ public void onUserFunctionStart(UserFunctionStartInfo info) {
// Still set the operation span as current so auto-instrumented calls become children
var operationSpan = operationSpans.get(info.id());
if (operationSpan != null) {
- var scope = operationSpan.makeCurrent();
+ var scope = OtelPluginSupport.onOwnerThread(operationSpan.makeCurrent());
var key = attemptKey(info.id(), info.attempt());
attemptScopes.put(key, scope);
}
@@ -575,7 +629,7 @@ public void onUserFunctionStart(UserFunctionStartInfo info) {
attemptSpans.put(key, span);
// Make span current on this thread so auto-instrumented calls become children
- var scope = span.makeCurrent();
+ var scope = OtelPluginSupport.onOwnerThread(span.makeCurrent());
attemptScopes.put(key, scope);
// Inject trace context into MDC for log-trace correlation
@@ -586,16 +640,21 @@ public void onUserFunctionStart(UserFunctionStartInfo info) {
@Override
public void onUserFunctionEnd(UserFunctionEndInfo info) {
+ if (invocationSpan == null) return;
+
var key = attemptKey(info.id(), info.attempt());
- // A running user function can finish after invocation End; its scope still belongs to this thread.
+
+ // A late user-function End still closes the scope on the owning thread.
attemptScopes.close(key);
- if (!tracingEnabled) return;
// Clear span-level MDC after user function completes (keep trace_id for handler-level logs between steps)
if (enableMdc) {
MDC.remove(MdcSpanEnricher.MDC_SPAN_ID);
}
+ // A bounded fatal return can finish spans before this operation owner finally unwinds.
+ if (ended) return;
+
// CONTEXT operations don't have attempt spans — scope cleanup is all we need
if ("CONTEXT".equals(info.type())) {
return;
@@ -628,31 +687,20 @@ public void onUserFunctionEnd(UserFunctionEndInfo info) {
// ─── Helpers ─────────────────────────────────────────────────────────
- private boolean bindTracer() {
- if (tracer != null) {
- return true;
- }
- synchronized (this) {
- if (tracer != null) {
- return true;
- }
- var setup = OtelPluginSupport.tryResolveGlobalProvider(instrumentationName, "InvocationOtelPlugin");
- if (setup == null) {
- return false;
- }
- sdkTracerProvider = setup.sdkTracerProvider();
- tracer = setup.tracer();
- return true;
- }
+ /**
+ * True when this instance emits no telemetry: either the tracer could not be bound for this invocation, or the
+ * invocation has already ended and its spans are closed.
+ */
+ private boolean disabled() {
+ return invocationSpan == null || ended;
}
private void endOpenSpansChildFirst() {
- // Attempt spans are children of operation spans.
+ // Close scopes owned by this thread, retaining other owners' scopes for their matching end hooks.
attemptScopes.closeCurrentThread();
for (var span : attemptSpans.values()) {
span.end();
}
- attemptSpans.clear();
// End still-open operation spans with the STARTED status set in onOperationStart.
// A later invocation's onOperationEnd emits a continuation span with the real terminal status.
@@ -663,8 +711,8 @@ private void endOpenSpansChildFirst() {
span.end();
}
}
- operationSpans.clear();
- operationContexts.clear();
+ // Every span has ended. Only unfinished owners may retain this invocation's instance for scope cleanup;
+ // these registries are never recycled for another invocation.
}
/**
@@ -767,17 +815,11 @@ private void addInitialOperationLink(SpanBuilder spanBuilder, String operationId
}
private TraceFlags effectiveTraceFlags() {
- var invocation = invocationSpan;
- if (invocation != null) {
- return invocation.getSpanContext().getTraceFlags();
- }
- var trace = executionTrace;
- return trace != null ? trace.flags() : TraceFlags.getDefault();
+ return invocationSpan.getSpanContext().getTraceFlags();
}
private TraceState effectiveTraceState() {
- var invocation = invocationSpan;
- return invocation != null ? invocation.getSpanContext().getTraceState() : TraceState.getDefault();
+ return invocationSpan.getSpanContext().getTraceState();
}
private static boolean isTerminal(InvocationEndInfo info) {
diff --git a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/InvocationOtelPluginProvider.java b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/InvocationOtelPluginProvider.java
index 165ed88fcd..caa5a0c0a8 100644
--- a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/InvocationOtelPluginProvider.java
+++ b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/InvocationOtelPluginProvider.java
@@ -3,31 +3,33 @@
package software.amazon.lambda.durable.otel;
import software.amazon.lambda.durable.plugin.DurableExecutionPlugin;
+import software.amazon.lambda.durable.plugin.DurableExecutionPluginFactory;
import software.amazon.lambda.durable.plugin.DurableExecutionPluginProvider;
+import software.amazon.lambda.durable.plugin.InvocationInfo;
/**
* Dynamically loads {@link InvocationOtelPlugin} when {@code DURABLE_EXECUTION_PLUGINS} contains
* {@code otel-invocation}.
+ *
+ * The provider is itself the per-invocation factory: it holds the environment-lifetime state (the ADOT global
+ * provider binding, the ID generator) once and creates one plugin instance per invocation from it.
*/
public final class InvocationOtelPluginProvider implements DurableExecutionPluginProvider {
+ private final DurableExecutionPluginFactory factory = InvocationOtelPlugin.factory();
+
@Override
public String getName() {
return "otel-invocation";
}
@Override
- public int getApiVersion() {
- return API_VERSION;
- }
-
- @Override
- public Class extends DurableExecutionPlugin> getPluginType() {
- return InvocationOtelPlugin.class;
+ public DurableExecutionPlugin createPlugin(InvocationInfo invocationInfo) {
+ return factory.createPlugin(invocationInfo);
}
@Override
- public DurableExecutionPlugin createPlugin() {
- return new InvocationOtelPlugin();
+ public String getExclusiveGroup() {
+ return factory.getExclusiveGroup();
}
}
diff --git a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/OtelPluginConfig.java b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/OtelPluginConfig.java
index ed4ac9be58..8fc5b136ac 100644
--- a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/OtelPluginConfig.java
+++ b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/OtelPluginConfig.java
@@ -10,8 +10,8 @@
* mirrors the {@code OtelPluginConfig} object in the JavaScript SDK and the {@code OtelPluginConfig} dataclass in the
* Python SDK for cross-SDK parity.
*
- *
Construct via {@link #builder()} and pass to a plugin's {@code (SdkTracerProviderBuilder, OtelPluginConfig)}
- * constructor:
+ *
Construct via {@link #builder()} and pass to a plugin's {@code factory(SdkTracerProviderBuilder,
+ * OtelPluginConfig)}:
*
*
{@code
* var config = OtelPluginConfig.builder()
@@ -20,7 +20,7 @@
* .workflowSpanName("Workflow")
* .instrumentationName("my-scope")
* .build();
- * var plugin = new InvocationOtelPlugin(tracerProviderBuilder, config);
+ * var factory = InvocationOtelPlugin.factory(tracerProviderBuilder, config);
* }
*
* Defaults: {@code contextExtractor = new XRayContextExtractor()}, {@code enableMdc = true}, {@code workflowSpanName
diff --git a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/OtelPluginEnvironment.java b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/OtelPluginEnvironment.java
new file mode 100644
index 0000000000..0593a3927a
--- /dev/null
+++ b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/OtelPluginEnvironment.java
@@ -0,0 +1,146 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package software.amazon.lambda.durable.otel;
+
+import io.opentelemetry.sdk.trace.SdkTracerProviderBuilder;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.lambda.durable.util.ExceptionHelper;
+
+/**
+ * Everything the OTel plugins need that belongs to the execution environment rather than to one invocation.
+ *
+ *
A plugin instance now serves exactly one Lambda invocation, so the objects that must exist once per environment
+ * live here: the resolved {@link OtelPluginConfig}, the {@link DeterministicIdGenerator}, and — for an
+ * application-owned tracer provider — the built provider and its tracer. {@code InvocationOtelPlugin.factory(...)} and
+ * {@code ExecutionOtelPlugin.factory(...)} create one of these and hand the same instance to every plugin instance they
+ * create, so the provider is built (and its ID generator and sampler installed) once per environment rather than once
+ * per invocation.
+ *
+ *
On the ADOT Java agent path there is no provider to build here: the global provider is resolved on first use and
+ * then cached. An invocation that runs before the agent has finished initializing therefore disables telemetry for
+ * itself only, and the next invocation's instance resolves the provider again.
+ */
+final class OtelPluginEnvironment {
+
+ private static final Logger logger = LoggerFactory.getLogger(OtelPluginEnvironment.class);
+
+ private final OtelPluginConfig config;
+ private final DeterministicIdGenerator idGenerator;
+
+ /** The application-owned provider and tracer, or null on the Java agent path. */
+ private final OtelPluginSupport.ProviderSetup ownedSetup;
+
+ /**
+ * The global provider and tracer, once resolved. Environment-lifetime state shared by every invocation's instance,
+ * hence volatile; a lost race only resolves the same global provider twice.
+ */
+ private volatile OtelPluginSupport.ProviderSetup resolvedGlobalSetup;
+
+ /**
+ * @throws NullPointerException if the config is null. The check belongs here because every factory overload on both
+ * plugins reaches this constructor, and because the alternative is silence: the global-provider path only
+ * stores the config, so a null one would first be dereferenced when an invocation's plugin instance is built,
+ * where {@code PluginRunner} contains the failure. The function would then run without the telemetry it asked
+ * for, reporting one warning per invocation. Registration is where a caller can still act on it.
+ */
+ private OtelPluginEnvironment(
+ OtelPluginConfig config, DeterministicIdGenerator idGenerator, OtelPluginSupport.ProviderSetup ownedSetup) {
+ this.config = Objects.requireNonNull(config, "config must not be null");
+ this.idGenerator = idGenerator;
+ this.ownedSetup = ownedSetup;
+ }
+
+ /**
+ * Builds the application-owned provider once: wraps the builder's ID generator and sampler, builds the provider and
+ * gets the tracer. Every invocation's plugin instance then shares them.
+ */
+ static OtelPluginEnvironment forProviderBuilder(
+ SdkTracerProviderBuilder tracerProviderBuilder, OtelPluginConfig config) {
+ // Checked before anything is consumed. The constructor below checks it too, but by then this method has
+ // installed the ID generator and the sampler on the caller's builder and built a provider -- and a provider
+ // that fails validation is unreachable, so its span processors and their worker threads are never shut down.
+ Objects.requireNonNull(config, "config must not be null");
+ var idGenerator = DeterministicIdGenerator.installOn(tracerProviderBuilder);
+ // Wrap the configured sampler so durable spans use the execution's single precomputed decision.
+ DurableSampler.installOn(tracerProviderBuilder);
+ var sdkTracerProvider = tracerProviderBuilder.build();
+ var setup = new OtelPluginSupport.ProviderSetup(
+ sdkTracerProvider, sdkTracerProvider.get(config.instrumentationName()));
+ return new OtelPluginEnvironment(config, idGenerator, setup);
+ }
+
+ /** The Java agent path: the global provider is resolved lazily, when an invocation's instance first needs it. */
+ static OtelPluginEnvironment forGlobalProvider(OtelPluginConfig config) {
+ return new OtelPluginEnvironment(config, OtelPluginSupport.createDefaultIdGenerator(), null);
+ }
+
+ /** Nonfatal failed construction can flush completed spans; known JVM fatals propagate without added cleanup. */
+ @SuppressWarnings("removal")
+ T createPlugin(Supplier constructor) {
+ try {
+ return constructor.get();
+ } catch (Throwable failure) {
+ var originalCause = ExceptionHelper.unwrapAsyncFailure(failure);
+ if (originalCause instanceof VirtualMachineError fatal) throw fatal;
+ if (originalCause instanceof ThreadDeath fatal) throw fatal;
+ flushAfterNonfatalConstructionFailure(failure);
+ throw failure;
+ }
+ }
+
+ @SuppressWarnings("removal")
+ private void flushAfterNonfatalConstructionFailure(Throwable failure) {
+ try {
+ // Do not rebind or reach through an opaque provider. Use only a setup the constructor could already use.
+ var setup = ownedSetup != null ? ownedSetup : resolvedGlobalSetup;
+ if (setup == null || setup.sdkTracerProvider() == null) return;
+ var result = setup.sdkTracerProvider().forceFlush().join(5, TimeUnit.SECONDS);
+ if (!result.isSuccess())
+ logger.warn("OTel startup cleanup flush failed or timed out — some spans may be lost");
+ } catch (Throwable cleanupFailure) {
+ if (failure == cleanupFailure) return;
+ var cleanupCause = ExceptionHelper.unwrapAsyncFailure(cleanupFailure);
+ if (cleanupCause instanceof VirtualMachineError || cleanupCause instanceof ThreadDeath) {
+ cleanupCause.addSuppressed(failure);
+ throw (Error) cleanupCause;
+ }
+ // Nonfatal cleanup must not replace the original nonfatal construction failure.
+ failure.addSuppressed(cleanupFailure);
+ }
+ }
+
+ OtelPluginConfig config() {
+ return config;
+ }
+
+ DeterministicIdGenerator idGenerator() {
+ return idGenerator;
+ }
+
+ /**
+ * The provider and tracer one invocation's plugin instance should use, or {@code null} when telemetry must be
+ * disabled for that invocation because the agent's global provider is not available yet.
+ *
+ * @param pluginName the plugin name used in diagnostics
+ */
+ OtelPluginSupport.ProviderSetup bind(String pluginName) {
+ if (ownedSetup != null) {
+ return ownedSetup;
+ }
+ var alreadyResolved = resolvedGlobalSetup;
+ if (alreadyResolved != null) {
+ return alreadyResolved;
+ }
+ var setup = OtelPluginSupport.tryResolveGlobalProvider(config.instrumentationName(), pluginName);
+ if (setup != null) {
+ // Resolution succeeded, so it holds for the rest of this environment's life: cache it so later invocations
+ // neither re-resolve nor re-log it. A failure is not cached — that is what makes the retry per invocation.
+ resolvedGlobalSetup = setup;
+ }
+ return setup;
+ }
+}
diff --git a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/OtelPluginSupport.java b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/OtelPluginSupport.java
index 6ce572c6d9..7687fc2715 100644
--- a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/OtelPluginSupport.java
+++ b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/OtelPluginSupport.java
@@ -22,13 +22,12 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.lambda.durable.exception.DurableOperationException;
-import software.amazon.lambda.durable.execution.DurableExecutor;
import software.amazon.lambda.durable.plugin.OperationEndInfo;
-/** Shared provider setup, sampling resolution, and invocation cleanup support. */
+/** Shared utilities for the OTel plugins' ADOT Java agent SPI path. */
final class OtelPluginSupport {
- /** Always runs invocation cleanup without hiding an earlier failure or a later JVM fatal. */
+ /** Always attempt provider cleanup, retaining the approved plugin-local failure precedence. */
static void finishWithRootCleanup(Runnable finish, Runnable cleanup) {
Throwable primary = null;
try {
@@ -59,32 +58,24 @@ private static int endFailurePriority(Throwable failure) {
return 1;
}
- /** Applies the base lifecycle's scope restoration with the same cleanup failure precedence. */
- static void runInvocationEnd(Scope scope, Runnable end) {
- finishWithRootCleanup(end, () -> {
- if (scope != null) scope.close();
- });
- }
-
- /** Rejects an older core before a plugin builds a provider or activates thread-local context. */
- static void requireSameThreadInvocationHooks() {
- var message = "This OpenTelemetry plugin requires same-thread invocation hooks from the Durable Execution"
- + " core. Upgrade the core together with the plugin layer.";
- try {
- if (!DurableExecutor.supportsSameThreadInvocationHooks()) throw new IllegalStateException(message);
- } catch (NoSuchMethodError missingCapability) {
- throw new IllegalStateException(message, missingCapability);
- }
- }
-
private static final Logger logger = LoggerFactory.getLogger(OtelPluginSupport.class);
private OtelPluginSupport() {}
- /** A visible replacement sampler cannot consume our bridge; an opaque agent provider may still need it. */
- static boolean usesDurableSamplingBridge(SdkTracerProvider provider) {
- // This identifies only the cross-loader wire consumer; context-key ownership still requires instanceof.
- return provider == null || provider.getSampler().getClass().getName().equals(DurableSampler.class.getName());
+ /** Keeps finalization on another thread from consuming an operation owner's eventual scope cleanup. */
+ static Scope onOwnerThread(Scope scope) {
+ var owner = Thread.currentThread();
+ return new Scope() {
+ // Only the owner can read or mutate this flag; other threads return before accessing it.
+ private boolean closed;
+
+ @Override
+ public void close() {
+ if (Thread.currentThread() != owner || closed) return;
+ closed = true;
+ scope.close();
+ }
+ };
}
/** Only the same DurableSampler class copy can consume this loader's parent-context holder. */
@@ -92,6 +83,11 @@ static boolean usesLocalDurableSampler(SdkTracerProvider provider) {
return provider != null && provider.getSampler() instanceof DurableSampler;
}
+ /** A visible replacement sampler cannot consume our bridge; an opaque agent provider may still need it. */
+ static boolean usesDurableSamplingBridge(SdkTracerProvider provider) {
+ return provider == null || provider.getSampler().getClass().getName().equals(DurableSampler.class.getName());
+ }
+
/** Terminal backend status and exception details are independent telemetry signals. */
static void applyOperationStatus(Span span, OperationEndInfo info) {
applyOperationStatus(span, info, null);
@@ -176,7 +172,7 @@ static DeterministicIdGenerator createDefaultIdGenerator() {
* sampled span yields {@code RECORD_AND_SAMPLE}; an unsampled but recording span yields {@code RECORD_ONLY}
* (its spans still reach processors); only an unsampled, non-recording span yields {@code DROP};
* Application-owned provider: configured sampler, once. When the tracer provider is reachable (the
- * two-argument constructor path), its sampler is read directly and evaluated a single time with
+ * application-owned provider path), its sampler is read directly and evaluated a single time with
* {@code ROOT_CONTEXT} (so a parent-based sampler applies its root policy), the canonical trace ID, span
* name, and attributes, and its full result is returned;
* Java-agent path: defer to the installed sampler. When the provider is not visible
diff --git a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/XRayContextExtractor.java b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/XRayContextExtractor.java
index 05a1110ab5..b15c978393 100644
--- a/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/XRayContextExtractor.java
+++ b/otel-plugin/src/main/java/software/amazon/lambda/durable/otel/XRayContextExtractor.java
@@ -7,9 +7,11 @@
import java.util.regex.Pattern;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import software.amazon.lambda.durable.plugin.InvocationInfo;
/**
- * Extracts OTel trace context from the AWS X-Ray {@code _X_AMZN_TRACE_ID} environment variable.
+ * Extracts OTel trace context from the invocation-local X-Ray header, using ordinary Lambda carriers only when no
+ * runtime snapshot is available.
*
* This extractor parses the Lambda/X-Ray header and returns the trace ID in OTel format (32 hex chars) along with
* the parent span ID (16 hex chars). Plugins use it as a fallback parent for Invocation spans; the deterministic
@@ -28,8 +30,40 @@ public class XRayContextExtractor implements ContextExtractor {
private static final Pattern HEX_32 = Pattern.compile("[0-9a-f]{32}");
private static final Pattern HEX_16 = Pattern.compile("[0-9a-f]{16}");
+ // Scoped only around extraction so legacy no-argument overrides retain virtual dispatch.
+ private final ThreadLocal invocationTraceHeader = new ThreadLocal<>();
+
+ @Override
+ public ExtractedContext extract(InvocationInfo info) {
+ var previous = invocationTraceHeader.get();
+ invocationTraceHeader.set(runtimeHeader(info));
+ try {
+ return extract();
+ } finally {
+ if (previous == null) {
+ invocationTraceHeader.remove();
+ } else {
+ invocationTraceHeader.set(previous);
+ }
+ }
+ }
+
+ private static String runtimeHeader(InvocationInfo info) {
+ try {
+ return info.xRayTraceId();
+ } catch (NoSuchMethodError olderCore) {
+ // Preserve extractor compatibility when InvocationInfo predates the optional accessor.
+ return null;
+ }
+ }
+
@Override
public ExtractedContext extract() {
+ var invocationHeader = invocationTraceHeader.get();
+ // A captured snapshot is authoritative even when empty or malformed; only null permits legacy fallback.
+ if (invocationHeader != null) {
+ return parseHeader(invocationHeader);
+ }
// Try system property first — the Lambda runtime interface client updates this per invocation, so it reflects
// the current invocation and avoids the JVM's process-lifetime environment-variable caching.
var traceHeader = System.getProperty(XRAY_SYSTEM_PROPERTY);
@@ -42,6 +76,10 @@ public ExtractedContext extract() {
return null;
}
+ return parseHeader(traceHeader);
+ }
+
+ private static ExtractedContext parseHeader(String traceHeader) {
String root = null;
String parent = null;
String sampled = null;
diff --git a/otel-plugin/src/test/compatibility/PluginLayerCompatibilityProbe.java b/otel-plugin/src/test/compatibility/PluginLayerCompatibilityProbe.java
new file mode 100644
index 0000000000..096c7e9d96
--- /dev/null
+++ b/otel-plugin/src/test/compatibility/PluginLayerCompatibilityProbe.java
@@ -0,0 +1,76 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.SdkTracerProviderBuilder;
+import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
+import java.lang.reflect.Method;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.nio.file.Path;
+import java.time.Instant;
+import java.util.List;
+import java.util.ServiceLoader;
+import software.amazon.lambda.durable.plugin.*;
+
+public class PluginLayerCompatibilityProbe {
+ public static void main(String[] args) throws Exception {
+ var path = Path.of(args[0]);
+ var view = args[1];
+ var header = "Root=1-6955b900-123456789012345678901234;Parent=1234567890123456;Sampled=1";
+ var old = System.getProperty("com.amazonaws.xray.traceHeader");
+ try (var layer = new URLClassLoader(new URL[] {path.toUri().toURL()},
+ PluginLayerCompatibilityProbe.class.getClassLoader())) {
+ System.setProperty("com.amazonaws.xray.traceHeader", header);
+ var provider = ServiceLoader.load(DurableExecutionPluginProvider.class, layer).stream()
+ .map(ServiceLoader.Provider::get).filter(p -> p.getName().equals(view)).findFirst().orElseThrow();
+ if (provider.getApiVersion() != DurableExecutionPluginProvider.API_VERSION) throw new AssertionError("provider API");
+ var type = provider.getPluginType();
+ if (type.getClassLoader() != layer) throw new AssertionError("plugin did not load from separate layer");
+ if (provider.createPlugin() == null) throw new AssertionError("dynamic provider constructor");
+ var configType = layer.loadClass("software.amazon.lambda.durable.otel.OtelPluginConfig");
+ var configBuilder = configType.getMethod("builder").invoke(null);
+ configBuilder.getClass().getMethod("enableMdc", boolean.class).invoke(configBuilder, false);
+ var config = configBuilder.getClass().getMethod("build").invoke(configBuilder);
+ var exporter = InMemorySpanExporter.create();
+ var builder = SdkTracerProvider.builder().addSpanProcessor(SimpleSpanProcessor.create(exporter));
+ var plugin = (DurableExecutionPlugin) type.getConstructor(SdkTracerProviderBuilder.class, configType)
+ .newInstance(builder, config);
+ var runner = new PluginRunner(List.of(plugin));
+ runner.onInvocationStart(new InvocationInfo("request", "arn:exec", true, Instant.EPOCH));
+ runner.onInvocationEnd(new InvocationEndInfo("request", "arn:exec", true, InvocationStatus.SUCCEEDED, null));
+ var spans = exporter.getFinishedSpanItems();
+ if (spans.size() != 2 || spans.stream().anyMatch(s -> !s.getTraceId().equals("6955b900123456789012345678901234")))
+ throw new AssertionError("legacy tracing failed: " + spans);
+ type.getMethods(); // Public reflection must also work with an older core.
+ checkOptionalHeaderDispatch(runner, plugin, exporter, type);
+ System.out.println("PASS " + view + " core=" + DurableExecutionPlugin.class.getProtectionDomain()
+ .getCodeSource().getLocation() + " layer=" + path);
+ } finally {
+ if (old == null) System.clearProperty("com.amazonaws.xray.traceHeader");
+ else System.setProperty("com.amazonaws.xray.traceHeader", old);
+ }
+ }
+ private static void checkOptionalHeaderDispatch(PluginRunner runner, DurableExecutionPlugin plugin,
+ InMemorySpanExporter exporter, Class> type) throws Exception {
+ Method hook;
+ try {
+ hook = PluginRunner.class.getMethod("onInvocationStart", InvocationInfo.class, String.class);
+ } catch (NoSuchMethodException olderCore) {
+ return; // The older core retains the original hook and ordinary carriers.
+ }
+ boolean supportsHeader;
+ try {
+ type.getDeclaredMethod("onInvocationStart", InvocationInfo.class, String.class);
+ supportsHeader = true;
+ } catch (NoSuchMethodException olderPlugin) {
+ supportsHeader = false;
+ }
+ exporter.reset();
+ hook.invoke(runner, new InvocationInfo("request-2", "arn:exec", false, Instant.EPOCH),
+ "Root=1-6955b900-aaaaaaaaaaaaaaaaaaaaaaaa;Parent=abcdefabcdefabcd;Sampled=0");
+ runner.onInvocationEnd(new InvocationEndInfo("request-2", "arn:exec", false, InvocationStatus.SUCCEEDED, null));
+ if (exporter.getFinishedSpanItems().size() != (supportsHeader ? 0 : 2))
+ throw new AssertionError("Optional dispatch did not preserve new/legacy behavior");
+ }
+}
diff --git a/otel-plugin/src/test/compatibility/b1/InstalledApiProbe.java b/otel-plugin/src/test/compatibility/b1/InstalledApiProbe.java
index 26721887b4..95008149f4 100644
--- a/otel-plugin/src/test/compatibility/b1/InstalledApiProbe.java
+++ b/otel-plugin/src/test/compatibility/b1/InstalledApiProbe.java
@@ -2,8 +2,8 @@
// SPDX-License-Identifier: Apache-2.0
package software.amazon.lambda.durable.otel;
-import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.read.ListAppender;
import io.opentelemetry.api.GlobalOpenTelemetry;
@@ -17,7 +17,6 @@
import java.time.Duration;
import java.util.List;
import java.util.ServiceLoader;
-import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.LoggerFactory;
import software.amazon.lambda.durable.model.ExecutionStatus;
@@ -26,9 +25,8 @@
import software.amazon.lambda.durable.plugin.InvocationEndInfo;
import software.amazon.lambda.durable.plugin.InvocationInfo;
import software.amazon.lambda.durable.testing.LocalDurableTestRunner;
-import software.amazon.lambda.durable.testing.TestResult;
-/** Compiled against the actual released 2.2.1 SPI, then run unchanged in fresh matrix JVMs. */
+/** Compiled against the current 3.x factory API, then run in fresh JVMs with each visible OTel API. */
public final class InstalledApiProbe {
private static final AtomicInteger HEALTHY_CREATED = new AtomicInteger();
private static final AtomicInteger HEALTHY_STARTS = new AtomicInteger();
@@ -44,9 +42,8 @@ public static void main(String[] args) throws Exception {
logs.start();
root.addAppender(logs);
try {
- if (Boolean.parseBoolean(args[7])) assertOldCoreRejected(view);
- else exercise(view, Boolean.parseBoolean(args[5]), Boolean.parseBoolean(args[6]), logs);
- System.out.println("COMPAT_PASS " + view + " negative=" + args[5] + " api=" + args[2]
+ exercise(view, Boolean.parseBoolean(args[5]), logs);
+ System.out.println("COMPAT_PASS " + view + " api=" + args[2]
+ " core=" + args[0] + " plugin=" + args[1]);
} finally {
root.detachAppender(logs);
@@ -64,61 +61,20 @@ private static void verifyArtifacts(String[] args) throws Exception {
.map(ServiceLoader.Provider::get)
.filter(value -> value.getName().equals(args[4]))
.findFirst().orElseThrow();
- check(provider.getApiVersion() == DurableExecutionPluginProvider.API_VERSION, "released SPI version");
- checkSource(provider.getPluginType(), Path.of(args[1]));
+ checkSource(provider.getClass(), Path.of(args[1]));
+ checkSource(LocalDurableTestRunner.class, Path.of(args[6]));
}
- private static void assertOldCoreRejected(String view) throws Exception {
- var provider = ServiceLoader.load(DurableExecutionPluginProvider.class).stream()
- .map(ServiceLoader.Provider::get).filter(value -> value.getName().equals(view))
- .findFirst().orElseThrow();
- expectCoreRejection(provider::createPlugin);
- var constructors = provider.getPluginType().getConstructors();
- check(constructors.length == 4, "exercise every published plugin constructor");
- for (var constructor : constructors) {
- var types = constructor.getParameterTypes();
- var parameters = new Object[types.length];
- for (int i = 0; i < types.length; i++) {
- parameters[i] = types[i].getName().endsWith("OtelPluginConfig")
- ? types[i].getMethod("defaults").invoke(null) : SdkTracerProvider.builder();
- }
- expectCoreRejection(() -> constructor.newInstance(parameters));
- }
- var handlerCalls = new AtomicInteger();
- expectCoreRejection(() -> createRunner(handlerCalls, new AtomicInteger()));
- check(handlerCalls.get() == 0 && HEALTHY_STARTS.get() == 0 && HEALTHY_ENDS.get() == 0,
- "configuration rejection must precede handler execution and lifecycle hooks");
- System.out.println("CORE_LIFECYCLE_REJECTION_CONFIRMED " + view);
- }
-
- private static void expectCoreRejection(Callable> construction) {
- Throwable rejected = null;
- try {
- construction.call();
- } catch (Throwable failure) {
- rejected = failure;
- }
- check(rejected != null, "new plugin silently accepted the released core without its lifecycle guarantee");
- for (var failure = rejected; failure != null; failure = failure.getCause()) {
- if (failure instanceof IllegalStateException && failure.getMessage() != null
- && failure.getMessage().contains("requires same-thread invocation hooks")) return;
- }
- throw new AssertionError("missing explicit core compatibility diagnostic", rejected);
- }
-
- private static void exercise(String view, boolean negative, boolean compatible, ListAppender logs) {
+ private static void exercise(String view, boolean compatible, ListAppender logs) {
GlobalOpenTelemetry.resetForTest();
// Reproduce #763's documented visible-API skew, not a claim of a deployed agent test.
OtelPluginAutoConfigurationState.markInstalled();
var handlerCalls = new AtomicInteger();
var sideEffects = new AtomicInteger();
var runner = createRunner(handlerCalls, sideEffects);
- check(HEALTHY_CREATED.get() == 1, "existing SPI must create the healthy plugin once per configuration");
+ check(HEALTHY_CREATED.get() == 0, "configuration must not create an invocation-owned plugin");
var first = runner.run("compatibility-input");
- if (negative) {
- assertNegative(first, handlerCalls, sideEffects);
- return;
- }
+ check(HEALTHY_CREATED.get() == 1, "first invocation must create its healthy plugin");
check(first.getStatus() == ExecutionStatus.PENDING, "plugin failure must preserve first invocation");
check(HEALTHY_STARTS.get() == 1 && HEALTHY_ENDS.get() == 1 && handlerCalls.get() == 1,
"healthy plugin and handler must run");
@@ -143,18 +99,6 @@ private static LocalDurableTestRunner createRunner(
});
}
- private static void assertNegative(
- TestResult result, AtomicInteger handlerCalls, AtomicInteger sideEffects) {
- check(result.getStatus() == ExecutionStatus.FAILED, "old/old older API must reproduce customer failure");
- var failure = result.getError().orElseThrow();
- check(failure.errorType().endsWith("NoSuchMethodError")
- && failure.errorMessage().contains("GlobalOpenTelemetry.isSet"),
- "negative control must reproduce the exact unsupported API: " + failure);
- check(HEALTHY_STARTS.get() == 0 && handlerCalls.get() == 0 && sideEffects.get() == 0,
- "old linkage failure must precede the healthy start hook and handler");
- System.out.println("NEGATIVE_CONTROL_REPRODUCED NoSuchMethodError GlobalOpenTelemetry.isSet");
- }
-
private static void resumeAndCheck(LocalDurableTestRunner runner, String view, boolean compatible,
AtomicInteger handlerCalls, AtomicInteger sideEffects) {
var exporter = InMemorySpanExporter.create();
@@ -167,7 +111,7 @@ private static void resumeAndCheck(LocalDurableTestRunner runner
check(HEALTHY_STARTS.get() == 2 && HEALTHY_ENDS.get() == 2 && handlerCalls.get() == 2,
"healthy hooks and handler must remain active on resume");
check(sideEffects.get() == 1, "completed user step must not repeat on resume");
- check(HEALTHY_CREATED.get() == 1, "resume must preserve the existing 2.x plugin instance lifetime");
+ check(HEALTHY_CREATED.get() == 2, "resume must create a new invocation-owned healthy plugin");
var spans = exporter.getFinishedSpanItems();
if (compatible) check(spans.stream().anyMatch(span -> span.getName().equals("Workflow")),
"compatible global provider must export Workflow spans in " + view);
@@ -191,19 +135,13 @@ private static SdkTracerProvider registerLateGlobal(boolean compatible, InMemory
return tracing;
}
- /** A real old-SPI service provider, discovered alongside the actual released/candidate OTel provider. */
+ /** A real factory service provider, discovered alongside the current OTel provider. */
public static final class HealthyProvider implements DurableExecutionPluginProvider {
@Override
public String getName() { return "compat-healthy"; }
@Override
- public int getApiVersion() { return API_VERSION; }
-
- @Override
- public Class extends DurableExecutionPlugin> getPluginType() { return HealthyPlugin.class; }
-
- @Override
- public DurableExecutionPlugin createPlugin() {
+ public DurableExecutionPlugin createPlugin(InvocationInfo info) {
HEALTHY_CREATED.incrementAndGet();
return new HealthyPlugin();
}
diff --git a/otel-plugin/src/test/compatibility/b1/README.md b/otel-plugin/src/test/compatibility/b1/README.md
index a0d8ab922f..b3d21b06e5 100644
--- a/otel-plugin/src/test/compatibility/b1/README.md
+++ b/otel-plugin/src/test/compatibility/b1/README.md
@@ -1,39 +1,33 @@
# Installed OpenTelemetry API compatibility
-Run after building the SDK and OTel plugin:
+Run after building the current core, testing SDK, and OTel plugin:
```sh
python3 .github/scripts/verify_otel_api_compatibility.py
```
-The driver resolves real released core, testing, and plugin **2.2.1** artifacts
-through Maven. It compiles one probe against the released SPI and runs fresh JVMs
-with old/old, new/old, old/new, and new/new core/plugin pairs, in both OTel views,
-using actual API/context **1.49.0** and **1.66.0** jars. Candidate jars come from
-the reactor build, or explicit `--new-core` / `--new-plugin` paths.
+This 3.x branch compiles its service-provider probe against the current factory
+API. It runs the current core, plugin, and testing SDK together in fresh JVMs,
+in both OTel views, with actual OpenTelemetry API/context **1.49.0** and **1.66.0**.
+The four required cases check:
-The sixteen required cases verify:
+- Core, plugin, testing SDK, API, and context classes load from the intended jars.
+- ServiceLoader discovers real providers; configuration creates no invocation
+ instances, and each initial/resumed invocation creates its own healthy plugin.
+- An unsupported visible API disables the affected instrumentation with a
+ diagnostic while healthy hooks, handler output, and completed-step replay work.
+- Early global-provider lookup does not install a no-op global. Later registration
+ succeeds, and API 1.66 exports real Workflow spans after resume.
-- Selected core, plugin, API, and context classes load from the intended jars.
-- The existing service-provider API discovers and creates a healthy plugin, with
- the same instance lifetime across suspension/resume.
-- Old/old plus API 1.49 reproduces the exact `GlobalOpenTelemetry.isSet`
- `NoSuchMethodError` and customer failure, before healthy start hooks/user code.
-- Fixed combinations isolate that mismatch, report a diagnostic, and preserve
- healthy hooks, handler output, and completed-step replay behavior.
-- An early unsupported/uninitialized global provider does not install a no-op
- global: subsequent registration must succeed. API 1.66 then exports real
- Workflow spans; unsupported API 1.49 disables the affected instrumentation.
+The independent fixture POM resolves the common dependency graph from released
+2.2.1 artifacts. The driver **replaces all three durable artifacts** with the
+current reactor jars before compiling or running. Explicit `--new-core`,
+`--new-plugin`, and `--new-testing` paths are available for an identical built tree.
+No 2.x/3.x interoperability is claimed: 3.x requires rebuilt factory providers.
+The 2.x branch retains its separate sixteen-case old/new compatibility matrix.
-Resolution, compilation, timeout, and probe failures all fail the command. There
-is no network-dependent skip. Logs, artifact SHA-256 hashes, negative-control
-results, and the case summary are written to `target/otel-api-compatibility`.
-The fixture POM is independent of the reactor and changes no production version
-or dependency floor.
-
-This matrix reproduces visible-API/classpath skew with real artifacts. Its test
-marker enables the documented plugin auto-configuration path; it does **not**
-claim to deploy or validate every Java-agent version. Actual agent validation
-must identify the released agent version, extension jar, visible API, runtime,
-and observed behavior separately. Provider registration remains fail-fast for
-invalid configuration; the fixture does not introduce invocation factories.
+Resolution, compilation, timeout, and probe failures fail the command; there is
+no network-dependent skip. Logs, jar hashes, and results go to
+`target/otel-api-compatibility`. No production dependencies or versions change.
+This is a visible-API test with an explicit auto-configuration marker, not a
+claim of deployed Java-agent coverage.
diff --git a/otel-plugin/src/test/compatibility/run_matrix.py b/otel-plugin/src/test/compatibility/run_matrix.py
new file mode 100644
index 0000000000..d38b5d629e
--- /dev/null
+++ b/otel-plugin/src/test/compatibility/run_matrix.py
@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Run 2.x-only installed core/plugin compatibility with an actual separate layer loader and a legacy-compiled caller.
+
+No downloads or AWS calls. Supply released core/plugin jars, built current artifacts/directories, and an existing
+classpath file containing their common dependencies. No SDK production dependencies are added for this check.
+"""
+import argparse
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import zipfile
+import xml.etree.ElementTree as ET
+
+parser = argparse.ArgumentParser(description=__doc__)
+for name in ("old-core", "old-plugin", "new-core", "new-plugin", "dependencies-classpath"):
+ parser.add_argument("--" + name, required=True, type=Path)
+parser.add_argument("--java-home", required=True, type=Path)
+args = parser.parse_args()
+def artifact_version(path):
+ if path.is_dir():
+ version_file = path / "version.prop"
+ if version_file.exists():
+ properties = version_file.read_text()
+ else:
+ pom = path.parent.parent / "pom.xml"
+ if not pom.exists():
+ parser.error(f"Cannot verify SDK version for {path}; supply a Maven artifact")
+ root = ET.parse(pom).getroot()
+ ns = {"m": "http://maven.apache.org/POM/4.0.0"}
+ version = root.find("m:version", ns)
+ if version is None:
+ version = root.find("m:parent/m:version", ns)
+ return version.text if version is not None else None
+ else:
+ with zipfile.ZipFile(path) as jar:
+ names = [name for name in jar.namelist()
+ if name.startswith("META-INF/maven/software.amazon.lambda.durable/")
+ and name.endswith("/pom.properties")]
+ if not names:
+ parser.error(f"Cannot verify SDK version for {path}")
+ properties = jar.read(names[0]).decode()
+ for line in properties.splitlines():
+ if line.startswith("version="):
+ return line.split("=", 1)[1].strip()
+ return None
+
+for artifact in (args.old_core, args.old_plugin, args.new_core, args.new_plugin):
+ version = artifact_version(artifact)
+ if not version or not version.startswith("2."):
+ parser.error(f"This legacy ABI harness accepts only 2.x artifacts; {artifact} has version {version!r}. "
+ "Use the factory/provider migration tests for 3.x.")
+
+known = {p.resolve() for p in (args.old_core, args.old_plugin, args.new_core, args.new_plugin)}
+entries = args.dependencies_classpath.read_text().strip().split(os.pathsep)
+deps = [p for p in entries if Path(p).resolve() not in known
+ and not Path(p).name.startswith("aws-durable-execution-sdk-java")]
+source = Path(__file__).with_name("PluginLayerCompatibilityProbe.java")
+target = Path.cwd() / "target"
+target.mkdir(exist_ok=True)
+with tempfile.TemporaryDirectory(prefix="layer-compat-", dir=target) as output:
+ subprocess.run([str(args.java_home / "bin/javac"), "--release", "17", "-cp",
+ os.pathsep.join([str(args.old_core), *deps]), "-d", output, str(source)], check=True)
+ for label, core, plugin in [("old/old", args.old_core, args.old_plugin),
+ ("old/new", args.old_core, args.new_plugin),
+ ("new/old", args.new_core, args.old_plugin),
+ ("new/new", args.new_core, args.new_plugin)]:
+ for view in ("otel-invocation", "otel-execution"):
+ result = subprocess.run([str(args.java_home / "bin/java"), "-cp",
+ os.pathsep.join([output, str(core), *deps]), "PluginLayerCompatibilityProbe", str(plugin), view],
+ capture_output=True, text=True)
+ if result.returncode:
+ raise RuntimeError(f"{label} {view} failed:\n{result.stdout}{result.stderr}")
+ print(label, result.stdout.strip())
diff --git a/otel-plugin/src/test/java/software/amazon/lambda/durable/OtelViewRegistrationTest.java b/otel-plugin/src/test/java/software/amazon/lambda/durable/OtelViewRegistrationTest.java
index b90217c375..58494dc79d 100644
--- a/otel-plugin/src/test/java/software/amazon/lambda/durable/OtelViewRegistrationTest.java
+++ b/otel-plugin/src/test/java/software/amazon/lambda/durable/OtelViewRegistrationTest.java
@@ -12,8 +12,6 @@
import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter;
import io.opentelemetry.sdk.trace.SdkTracerProvider;
import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
-import java.io.File;
-import java.net.URISyntaxException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Duration;
@@ -24,8 +22,8 @@
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.MDC;
-import software.amazon.lambda.durable.context.DurableContextImpl;
import software.amazon.lambda.durable.model.ExecutionStatus;
import software.amazon.lambda.durable.otel.ExecutionOtelPlugin;
import software.amazon.lambda.durable.otel.ExecutionOtelPluginProvider;
@@ -33,53 +31,33 @@
import software.amazon.lambda.durable.otel.InvocationOtelPluginProvider;
import software.amazon.lambda.durable.otel.OtelPluginConfig;
import software.amazon.lambda.durable.plugin.DurableExecutionPlugin;
-import software.amazon.lambda.durable.plugin.ExclusivePluginGroup;
+import software.amazon.lambda.durable.plugin.DurableExecutionPluginFactory;
+import software.amazon.lambda.durable.plugin.InvocationInfo;
import software.amazon.lambda.durable.testing.LocalDurableTestRunner;
class OtelViewRegistrationTest {
@Test
- void existingSubclassGroupMethodsRemainCompatible() {
- // Downstream subclasses could already declare this method before group metadata joined the plugin API.
- var invocation = new InvocationOtelPlugin() {
+ void factoryMetadataRejectsConflictsBeforeCreatingInstances() {
+ var created = new AtomicInteger();
+ var factory = new DurableExecutionPluginFactory() {
+ @Override
public String getExclusiveGroup() {
- return "legacy application group";
+ return "custom-tracing";
}
- };
- var execution = new ExecutionOtelPlugin() {
- public String getExclusiveGroup() {
- return null;
+
+ @Override
+ public DurableExecutionPlugin createPlugin(InvocationInfo info) {
+ created.incrementAndGet();
+ return new DurableExecutionPlugin() {};
}
};
- assertEquals("legacy application group", invocation.getExclusiveGroup());
- assertNull(execution.getExclusiveGroup());
- var error = assertThrows(
- IllegalStateException.class,
- () -> DurableConfig.builder().withPlugins(invocation, execution).build());
- assertTrue(error.getMessage().contains(invocation.getClass().getName()));
- assertTrue(error.getMessage().contains(execution.getClass().getName()));
- assertDoesNotThrow(() -> DurableConfig.builder().withPlugins(invocation).build());
- }
-
- @ParameterizedTest
- @CsvSource({"true,false", "true,true", "false,false", "false,true"})
- void reannotatedSubclassesRetainViewExclusivity(boolean executionSubclass, boolean reversed) {
- DurableExecutionPlugin custom =
- executionSubclass ? new ReannotatedExecutionPlugin() : new ReannotatedInvocationPlugin();
- DurableExecutionPlugin opposite = executionSubclass ? new InvocationOtelPlugin() : new ExecutionOtelPlugin();
- var first = reversed ? opposite : custom;
- var second = reversed ? custom : opposite;
- var error = assertThrows(
- IllegalStateException.class,
- () -> DurableConfig.builder().withPlugins(first, second).build());
- assertTrue(error.getMessage().contains("durable-otel-view"));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> DurableConfig.builder().withPlugins(factory, factory).build());
+ assertDoesNotThrow(() -> DurableConfig.builder().withPlugins(factory).build());
+ assertEquals(0, created.get());
}
- @ExclusivePluginGroup("application-instrumentation")
- private static class ReannotatedInvocationPlugin extends InvocationOtelPlugin {}
-
- @ExclusivePluginGroup("application-instrumentation")
- private static class ReannotatedExecutionPlugin extends ExecutionOtelPlugin {}
-
@ParameterizedTest
@CsvSource({"explicit,false", "explicit,true", "dynamic,false", "dynamic,true", "mixed,false", "mixed,true"})
void rejectsBothViewsBeforeEmissionAndLeavesContextUntouched(String path, boolean reversed) {
@@ -89,9 +67,9 @@ void rejectsBothViewsBeforeEmissionAndLeavesContextUntouched(String path, boolea
var providers = List.of(new ExecutionOtelPluginProvider(), new InvocationOtelPluginProvider());
var names = reversed ? "otel-execution,otel-invocation" : "otel-invocation,otel-execution";
var plugins = switch (path) {
- case "dynamic" -> DynamicPluginLoader.loadConfiguredPlugins(names, providers, List.of());
+ case "dynamic" -> DynamicPluginLoader.loadConfiguredPluginFactories(names, providers, List.of());
case "mixed" ->
- DynamicPluginLoader.loadConfiguredPlugins(
+ DynamicPluginLoader.loadConfiguredPluginFactories(
reversed ? "otel-execution" : "otel-invocation", providers, List.of(second));
default -> List.of(first, second);
};
@@ -105,11 +83,10 @@ void rejectsBothViewsBeforeEmissionAndLeavesContextUntouched(String path, boolea
MDC.put("trace_id", "existing");
try {
var error = assertThrows(
- IllegalStateException.class,
+ IllegalArgumentException.class,
() -> DurableConfig.builder()
- .withPlugins(plugins.toArray(DurableExecutionPlugin[]::new))
+ .withPlugins(plugins.toArray(DurableExecutionPluginFactory[]::new))
.build());
- assertTrue(error.getMessage().startsWith("Dynamic plugin configuration failed: "));
assertTrue(error.getMessage().contains("InvocationOtelPlugin"));
assertTrue(error.getMessage().contains("ExecutionOtelPlugin"));
assertTrue(error.getMessage().contains("only one"));
@@ -122,30 +99,13 @@ void rejectsBothViewsBeforeEmissionAndLeavesContextUntouched(String path, boolea
}
}
- @ParameterizedTest
- @CsvSource({"true", "false"})
- void repeatedExplicitViewInstancesAreRejectedBeforeEmission(boolean executionView) {
- var exporter = InMemorySpanExporter.create();
- var first = plugin(executionView, exporter);
- var second = plugin(executionView, exporter);
- var ambient = Context.current();
- assertThrows(
- IllegalStateException.class,
- () -> DurableConfig.builder().withPlugins(first, second).build());
- assertThrows(
- IllegalStateException.class,
- () -> DurableConfig.builder().withPlugins(first, first).build());
- assertTrue(exporter.getFinishedSpanItems().isEmpty());
- assertSame(ambient, Context.current());
- }
-
@ParameterizedTest
@CsvSource({"true,true", "true,false", "false,true", "false,false"})
void singleViewWithUnrelatedPluginPreservesResumeAndOutcome(boolean executionView, boolean success) {
var exporter = InMemorySpanExporter.create();
var view = plugin(executionView, exporter);
var config = DurableConfig.builder()
- .withPlugins(view, new DurableExecutionPlugin() {})
+ .withPlugins(view, info -> new DurableExecutionPlugin() {})
.build();
var effects = new AtomicInteger();
var runner = LocalDurableTestRunner.create(
@@ -175,18 +135,13 @@ void singleViewWithUnrelatedPluginPreservesResumeAndOutcome(boolean executionVie
}
@ParameterizedTest
- @CsvSource({"otel-invocation,false", "otel-execution,false", "otel-invocation,true", "otel-execution,true"})
- void environmentSelectedSingleViewCanBeCopiedIntoLocalRunner(
- String provider, boolean releasedTestingSdk, @TempDir Path directory) throws Exception {
+ @ValueSource(strings = {"otel-invocation", "otel-execution"})
+ void environmentSelectedSingleViewCanBeCopiedIntoLocalRunner(String provider, @TempDir Path directory)
+ throws Exception {
var java = Path.of(System.getProperty("java.home"), "bin", "java").toString();
var classpath = System.getProperty("surefire.test.class.path", System.getProperty("java.class.path"));
- var testingLocation =
- releasedTestingSdk ? Path.of(System.getProperty("releasedTestingSdkJar")) : testingSdkLocation();
- assertTrue(Files.exists(testingLocation), "Testing SDK artifact must exist");
- classpath = testingLocation + File.pathSeparator + classpath;
var output = directory.resolve("child.log");
- var builder = new ProcessBuilder(
- java, "-cp", classpath, EnvironmentRunnerCheck.class.getName(), testingLocation.toString())
+ var builder = new ProcessBuilder(java, "-cp", classpath, EnvironmentRunnerCheck.class.getName())
.redirectErrorStream(true)
.redirectOutput(output.toFile());
builder.environment().put("DURABLE_EXECUTION_PLUGINS", provider);
@@ -199,14 +154,6 @@ void environmentSelectedSingleViewCanBeCopiedIntoLocalRunner(
}
}
- private static Path testingSdkLocation() throws URISyntaxException {
- return Path.of(LocalDurableTestRunner.class
- .getProtectionDomain()
- .getCodeSource()
- .getLocation()
- .toURI());
- }
-
private static String read(Path path) {
try {
return Files.readString(path);
@@ -216,78 +163,46 @@ private static String read(Path path) {
}
public static class EnvironmentRunnerCheck {
- public static void main(String[] args) throws Exception {
- // Ensure the old-testing cases execute the released bytecode, not the reactor's updated runner.
- assertEquals(Path.of(args[0]).toRealPath(), testingSdkLocation().toRealPath());
+ public static void main(String[] args) {
var config = DurableConfig.builder()
.withDeserializeAfterSerialization(false)
.build();
- assertEquals(1, config.getPluginRunner().getPlugins().size());
+ assertEquals(1, config.getPluginFactories().size());
var copy = config.toBuilder().build();
- assertEquals(
- config.getPluginRunner().getPlugins(),
- copy.getPluginRunner().getPlugins());
+ assertEquals(config.getPluginFactories(), copy.getPluginFactories());
assertFalse(copy.shouldDeserializeAfterSerialization());
- assertCopiedBuilderReplacementDoesNotRediscover(config);
- var effects = new AtomicInteger();
+ assertTrue(config.toBuilder()
+ .withPlugins()
+ .build()
+ .getPluginFactories()
+ .isEmpty());
+ DurableExecutionPluginFactory replacement = info -> new DurableExecutionPlugin() {};
+ assertEquals(
+ List.of(replacement),
+ config.toBuilder().withPlugins(replacement).build().getPluginFactories());
+ assertEquals(1, DurableConfig.builder().build().getPluginFactories().size());
var runner = LocalDurableTestRunner.create(
String.class,
(input, ctx) -> {
- var effective = ((DurableContextImpl) ctx)
- .getDurableConfig()
- .getPluginRunner()
- .getPlugins();
- assertEquals(1, effective.size(), "Released and current runner copies must retain one view");
- assertSame(config.getPluginRunner().getPlugins().get(0), effective.get(0));
- ctx.step("once", Integer.class, step -> effects.incrementAndGet());
ctx.wait("pause", Duration.ofSeconds(1));
return input;
},
config);
assertEquals(ExecutionStatus.PENDING, runner.run("input").getStatus());
runner.advanceTime();
- var result = runner.runUntilComplete("input");
- assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
- assertEquals("input", result.getResult(String.class));
- assertEquals(1, effects.get());
- }
-
- private static void assertCopiedBuilderReplacementDoesNotRediscover(DurableConfig config) {
- var replacement = new DurableExecutionPlugin() {};
- var copiedBuilder = config.toBuilder();
- assertTrue(copiedBuilder
- .withPlugins()
- .build()
- .getPluginRunner()
- .getPlugins()
- .isEmpty());
- assertEquals(
- List.of(replacement),
- copiedBuilder
- .withPlugins(replacement)
- .build()
- .getPluginRunner()
- .getPlugins());
-
- var fresh = DurableConfig.builder()
- .withPlugins(replacement)
- .build()
- .getPluginRunner()
- .getPlugins();
- assertEquals(2, fresh.size(), "Fresh builders continue to honor environment discovery");
assertEquals(
- config.getPluginRunner().getPlugins().get(0).getClass(),
- fresh.get(0).getClass());
- assertSame(replacement, fresh.get(1));
+ ExecutionStatus.SUCCEEDED, runner.runUntilComplete("input").getStatus());
}
}
- private static DurableExecutionPlugin plugin(boolean executionView, InMemorySpanExporter exporter) {
+ private static DurableExecutionPluginFactory plugin(boolean executionView, InMemorySpanExporter exporter) {
var builder = SdkTracerProvider.builder().addSpanProcessor(SimpleSpanProcessor.create(exporter));
var config = OtelPluginConfig.builder()
.contextExtractor(() -> null)
.enableMdc(false)
.build();
- return executionView ? new ExecutionOtelPlugin(builder, config) : new InvocationOtelPlugin(builder, config);
+ return executionView
+ ? ExecutionOtelPlugin.factory(builder, config)
+ : InvocationOtelPlugin.factory(builder, config);
}
}
diff --git a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/BackendOperationTimingTest.java b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/BackendOperationTimingTest.java
index 5c91d0413b..328dc35cf7 100644
--- a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/BackendOperationTimingTest.java
+++ b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/BackendOperationTimingTest.java
@@ -29,6 +29,7 @@
import software.amazon.lambda.durable.model.DurableExecutionInput;
import software.amazon.lambda.durable.model.ExecutionStatus;
import software.amazon.lambda.durable.plugin.DurableExecutionPlugin;
+import software.amazon.lambda.durable.plugin.DurableExecutionPluginFactory;
import software.amazon.lambda.durable.plugin.OperationEndInfo;
@Timeout(15)
@@ -98,9 +99,9 @@ public GetDurableExecutionStateResponse getExecutionState(String arn, String tok
.enableMdc(false)
.contextExtractor(() -> null)
.build();
- DurableExecutionPlugin otel = executionView
- ? new ExecutionOtelPlugin(builder, settings)
- : new InvocationOtelPlugin(builder, settings);
+ DurableExecutionPluginFactory otel = executionView
+ ? ExecutionOtelPlugin.factory(builder, settings)
+ : InvocationOtelPlugin.factory(builder, settings);
var ends = new CopyOnWriteArrayList();
var observer = new DurableExecutionPlugin() {
public void onOperationEnd(OperationEndInfo info) {
@@ -110,7 +111,7 @@ public void onOperationEnd(OperationEndInfo info) {
var caught = new AtomicInteger();
var config = DurableConfig.builder()
.withDurableExecutionClient(client)
- .withPlugins(otel, observer)
+ .withPlugins(otel, ignored -> observer)
.build();
for (int invocation = 0; invocation < 2; invocation++) {
var firstLive = live && invocation == 0;
diff --git a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/CallbackErrorMetadataTest.java b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/CallbackErrorMetadataTest.java
index b79d439ed5..546e55611a 100644
--- a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/CallbackErrorMetadataTest.java
+++ b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/CallbackErrorMetadataTest.java
@@ -21,6 +21,7 @@
import software.amazon.lambda.durable.exception.DurableOperationException;
import software.amazon.lambda.durable.model.ExecutionStatus;
import software.amazon.lambda.durable.plugin.DurableExecutionPlugin;
+import software.amazon.lambda.durable.plugin.DurableExecutionPluginFactory;
import software.amazon.lambda.durable.plugin.InvocationEndInfo;
import software.amazon.lambda.durable.plugin.InvocationInfo;
import software.amazon.lambda.durable.plugin.OperationEndInfo;
@@ -106,9 +107,9 @@ void publicCallbackPreservesStoredFailureButReportsOnlyPresentErrorDetails(
.contextExtractor(() -> null)
.enableMdc(false)
.build();
- DurableExecutionPlugin otel = executionView
- ? new ExecutionOtelPlugin(provider, options)
- : new InvocationOtelPlugin(provider, options);
+ DurableExecutionPluginFactory otel = executionView
+ ? ExecutionOtelPlugin.factory(provider, options)
+ : InvocationOtelPlugin.factory(provider, options);
var starts = new CopyOnWriteArrayList();
var ends = new CopyOnWriteArrayList();
var operationEnds = new CopyOnWriteArrayList();
@@ -133,7 +134,9 @@ public void onOperationEnd(OperationEndInfo info) {
String.class,
(input, context) -> context.waitForCallback(
"external-failure", String.class, (callbackId, step) -> submitterCalls.incrementAndGet()),
- DurableConfig.builder().withPlugins(otel, observer).build());
+ DurableConfig.builder()
+ .withPlugins(otel, ignored -> observer)
+ .build());
assertEquals(ExecutionStatus.PENDING, runner.run("input").getStatus());
var callbackId = runner.getCallbackId(CALLBACK);
if (timeout) runner.timeoutCallback(callbackId);
diff --git a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/CaughtCallbackFailureStatusTest.java b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/CaughtCallbackFailureStatusTest.java
index ffdf7c98d1..3198a48e44 100644
--- a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/CaughtCallbackFailureStatusTest.java
+++ b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/CaughtCallbackFailureStatusTest.java
@@ -26,6 +26,7 @@
import software.amazon.lambda.durable.exception.DurableOperationException;
import software.amazon.lambda.durable.model.ExecutionStatus;
import software.amazon.lambda.durable.plugin.DurableExecutionPlugin;
+import software.amazon.lambda.durable.plugin.DurableExecutionPluginFactory;
import software.amazon.lambda.durable.plugin.OperationEndInfo;
import software.amazon.lambda.durable.testing.LocalDurableTestRunner;
@@ -59,9 +60,9 @@ void caughtCallbackFailureIsErrorWhileHandlerSucceedsAndReplaySkipsSavedEffects(
.enableMdc(false)
.contextExtractor(() -> null)
.build();
- DurableExecutionPlugin otel = executionView
- ? new ExecutionOtelPlugin(builder, settings)
- : new InvocationOtelPlugin(builder, settings);
+ DurableExecutionPluginFactory otel = executionView
+ ? ExecutionOtelPlugin.factory(builder, settings)
+ : InvocationOtelPlugin.factory(builder, settings);
var ends = new CopyOnWriteArrayList();
var observer = new DurableExecutionPlugin() {
@Override
@@ -92,7 +93,9 @@ public void onOperationEnd(OperationEndInfo info) {
context.wait("resume-once", Duration.ofSeconds(1));
return "handled";
},
- DurableConfig.builder().withPlugins(otel, observer).build());
+ DurableConfig.builder()
+ .withPlugins(otel, ignored -> observer)
+ .build());
runnerRef.set(runner);
assertEquals(ExecutionStatus.PENDING, runner.run("input").getStatus());
if (!live) {
diff --git a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/CaughtFailureConformanceHandlerTest.java b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/CaughtFailureConformanceHandlerTest.java
index cfbc85c2eb..fc9a120741 100644
--- a/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/CaughtFailureConformanceHandlerTest.java
+++ b/otel-plugin/src/test/java/software/amazon/lambda/durable/otel/CaughtFailureConformanceHandlerTest.java
@@ -414,9 +414,9 @@ private Fixture(boolean execution, Path classes, String handlerName) throws Exce
.enableMdc(false)
.contextExtractor(() -> null)
.build();
- DurableExecutionPlugin otel = execution
- ? new ExecutionOtelPlugin(provider, options)
- : new InvocationOtelPlugin(provider, options);
+ DurableExecutionPluginFactory otel = execution
+ ? ExecutionOtelPlugin.factory(provider, options)
+ : InvocationOtelPlugin.factory(provider, options);
var observer = new DurableExecutionPlugin() {
public void onOperationEnd(OperationEndInfo info) {
operationEnds.add(info);
@@ -443,7 +443,7 @@ public void onUserFunctionEnd(UserFunctionEndInfo info) {
};
config = DurableConfig.builder()
.withExecutorService(workers)
- .withPlugins(otel, observer)
+ .withPlugins(otel, ignored -> observer)
.build();
runner = LocalDurableTestRunner.create(
new TypeToken