refactor(plugin)!: create one plugin instance per invocation

.github/scripts/verify_otel_api_compatibility.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
-"""Exercise real released/candidate core and plugin artifacts with two visible OTel APIs.
+"""Exercise the current 3.x core, plugin, and testing artifacts with two visible OTel APIs.
 
 Dependency resolution and every probe are required: a network, compilation, or case
 failure returns nonzero. No production dependency versions are modified.
@@ -105,72 +105,59 @@ def run_matrix(args: argparse.Namespace) -> int:
     output.mkdir(parents=True, exist_ok=True)
     fixture = root / "otel-plugin/src/test/compatibility/b1"
     cp = resolve_classpaths(fixture, output, args.maven)
-    released_core = artifact(cp["1.66.0"], CORE, RELEASED_VERSION)
-    released_plugin = artifact(cp["1.66.0"], PLUGIN, RELEASED_VERSION)
-    new_core = args.new_core.resolve() if args.new_core else candidate_jar(root, "sdk", CORE)
-    new_plugin = args.new_plugin.resolve() if args.new_plugin else candidate_jar(root, "otel-plugin", PLUGIN)
-    for jar in (new_core, new_plugin):
-        if not jar.is_file():
-            raise RuntimeError(f"Candidate artifact missing: {jar}")
-    candidate_inputs = {"core": jar_facts(new_core), "plugin": jar_facts(new_plugin)}
-    new_core = snapshot_candidate(new_core, output)
-    new_plugin = snapshot_candidate(new_plugin, output)
+    testing_name = CORE + "-testing"
+    inputs = {
+        "core": args.new_core.resolve() if args.new_core else candidate_jar(root, "sdk", CORE),
+        "plugin": args.new_plugin.resolve() if args.new_plugin else candidate_jar(root, "otel-plugin", PLUGIN),
+        "testing": args.new_testing.resolve() if args.new_testing else candidate_jar(root, "sdk-testing", testing_name),
+    }
+    candidate_inputs = {name: jar_facts(path) for name, path in inputs.items()}
+    selected = {name: snapshot_candidate(path, output) for name, path in inputs.items()}
+    excluded = {f"{name}-{RELEASED_VERSION}.jar" for name in (CORE, PLUGIN, testing_name)}
+    dependencies = {version: [path for path in paths if path.name not in excluded]
+                    for version, paths in cp.items()}
     classes = output / "classes"
     classes.mkdir(exist_ok=True)
-    execute([args.javac, "--release", "17", "-classpath", os.pathsep.join(map(str, cp["1.66.0"])),
+    compile_cp = [*selected.values(), *dependencies["1.66.0"]]
+    execute([args.javac, "--release", "17", "-classpath", os.pathsep.join(map(str, compile_cp)),
              "-d", str(classes), str(fixture / "InstalledApiProbe.java")], output / "compile.log")
     services = classes / "META-INF/services"
     services.mkdir(parents=True, exist_ok=True)
     (services / "software.amazon.lambda.durable.plugin.DurableExecutionPluginProvider").write_text(
         PROBE + "$HealthyProvider\n")
     report: dict[str, object] = {
-        "released_core": jar_facts(released_core), "released_plugin": jar_facts(released_plugin),
-        "new_core": jar_facts(new_core), "new_plugin": jar_facts(new_plugin),
+        "contract": "Current 3.x factory API; cross-major core/plugin mixtures are unsupported.",
         "candidate_inputs": candidate_inputs,
+        "candidate_snapshots": {name: jar_facts(path) for name, path in selected.items()},
         "cases": [], "agent_coverage": "This matrix is visible-API skew, not a deployed Java-agent test.",
     }
     cases: list[dict[str, object]] = report["cases"]  # type: ignore[assignment]
     failures = 0
-    pairs = {"old-old": (released_core, released_plugin), "new-old": (new_core, released_plugin),
-             "old-new": (released_core, new_plugin), "new-new": (new_core, new_plugin)}
     for version in API_VERSIONS:
         api = artifact(cp[version], "opentelemetry-api", version)
         context = artifact(cp[version], "opentelemetry-context", version)
-        dependencies = [p for p in cp[version] if p.name not in
-                        (f"{CORE}-{RELEASED_VERSION}.jar", f"{PLUGIN}-{RELEASED_VERSION}.jar")]
-        for label, (core, plugin) in pairs.items():
-            for view in ("otel-invocation", "otel-execution"):
-                name = f"{label}-api{version}-{view}"
-                negative = label == "old-old" and version == "1.49.0"
-                incompatible_core = label == "old-new"
-                case: dict[str, object] = {"name": name, "expected_negative_control": negative,
-                                          "expected_core_rejection": incompatible_core,
-                                          "api": jar_facts(api), "context": jar_facts(context)}
-                command = [args.java, "-cp", os.pathsep.join(map(str, [classes, core, plugin, *dependencies])),
-                           PROBE, str(core), str(plugin), str(api), str(context), view,
-                           str(negative).lower(), str(version == "1.66.0").lower(),
-                           str(incompatible_core).lower()]
-                try:
-                    log = output / f"{name}.log"
-                    execute(command, log, env=probe_environment(view), timeout=90)
-                    contents = log.read_text(errors="replace")
-                    if "COMPAT_PASS " not in contents:
-                        raise RuntimeError("Probe did not report successful completion")
-                    if negative and "NEGATIVE_CONTROL_REPRODUCED" not in contents:
-                        raise RuntimeError("Released negative control did not reproduce the reported failure")
-                    if incompatible_core and "CORE_LIFECYCLE_REJECTION_CONFIRMED" not in contents:
-                        raise RuntimeError("Old core must reject the new plugin before invocation startup")
-                    case["passed"] = True
-                except RuntimeError as error:
-                    failures += 1
-                    case.update(passed=False, error=str(error))
-                cases.append(case)
-                (output / "results.json").write_text(json.dumps(report, indent=2) + "\n")
-                print(f"{'PASS' if case['passed'] else 'FAIL'} {name}", flush=True)
+        for view in ("otel-invocation", "otel-execution"):
+            name = f"current3x-api{version}-{view}"
+            case: dict[str, object] = {"name": name, "api": jar_facts(api), "context": jar_facts(context)}
+            command = [args.java, "-cp", os.pathsep.join(map(str, [classes, *selected.values(), *dependencies[version]])),
+                       PROBE, str(selected["core"]), str(selected["plugin"]), str(api), str(context), view,
+                       str(version == "1.66.0").lower(), str(selected["testing"])]
+            try:
+                log = output / f"{name}.log"
+                execute(command, log, env=probe_environment(view), timeout=90)
+                if "COMPAT_PASS " not in log.read_text(errors="replace"):
+                    raise RuntimeError("Probe did not report successful completion")
+                case["passed"] = True
+            except RuntimeError as error:
+                failures += 1
+                case.update(passed=False, error=str(error))
+            cases.append(case)
+            (output / "results.json").write_text(json.dumps(report, indent=2) + "\n")
+            print(f"{'PASS' if case['passed'] else 'FAIL'} {name}", flush=True)
     report["passed"] = failures == 0
     report["failure_count"] = failures
     (output / "results.json").write_text(json.dumps(report, indent=2) + "\n")
-    print(f"Installed artifact matrix: {len(cases) - failures}/{len(cases)} passed; {output / 'results.json'}")
+    print(f"Current 3.x artifact matrix: {len(cases) - failures}/{len(cases)} passed; {output / 'results.json'}")
     return 1 if failures else 0
 
 
@@ -180,6 +167,7 @@ def main() -> int:
     parser.add_argument("--output", type=Path, default=Path("target/otel-api-compatibility"))
     parser.add_argument("--new-core", type=Path)
     parser.add_argument("--new-plugin", type=Path)
+    parser.add_argument("--new-testing", type=Path)
     parser.add_argument("--maven", default=shutil.which("mvn") or "mvn")
     parser.add_argument("--java", default=shutil.which("java") or "java")
     parser.add_argument("--javac", default=shutil.which("javac") or "javac")

.github/workflows/conformance-tests.yml

@@ -21,7 +21,7 @@ on:
 
 concurrency:
   # Runs share fixed per-suite stack names, so serialize the whole workflow
-  # (queue, don't cancel) to avoid concurrent CloudFormation updates on the
+  # with multiple pending runs to avoid concurrent CloudFormation updates on the
   # same stack -- mirrors e2e-tests.yml.
   group: conformance-tests
   cancel-in-progress: false

.github/workflows/e2e-tests.yml

@@ -24,6 +24,7 @@ on:
       - 'pom.xml'
 
 concurrency:
+  # Shared stacks require serial deployment; preserve waiting runs from other PRs.
   group: e2e-tests
   cancel-in-progress: false
   queue: max
@@ -149,6 +150,26 @@ jobs:
           if [[ "$invariant_violated" == true ]]; then
             exit 1
           fi
+      - name: Collect Lambda errors after failed E2E tests
+        if: failure() && env.E2E_LOG_START_TIME_MS != ''
+        env:
+          E2E_STACK_NAME: Java${{ matrix.java }}-JavaSDKCloudBasedIntegrationTestStack
+        run: |
+          set -euo pipefail
+          log_groups=$(aws cloudformation list-stack-resources \
+            --stack-name "$E2E_STACK_NAME" \
+            --query "StackResourceSummaries[?ResourceType=='AWS::Logs::LogGroup'].PhysicalResourceId" \
+            --output text)
+          for log_group in $log_groups; do
+            echo "::group::Lambda errors: $log_group"
+            aws logs filter-log-events \
+              --log-group-name "$log_group" \
+              --start-time "$E2E_LOG_START_TIME_MS" \
+              --filter-pattern '%ERROR|Error|Exception|timed.out|Invalid.suspension|not.active|already.registered%' \
+              --limit 100 --no-paginate \
+              --query 'events[].message' --output text
+            echo "::endgroup::"
+          done
       - name: Publish test case summary
         if: always()
         env:

.github/workflows/lmi-e2e-tests.yml

@@ -0,0 +1,128 @@
+name: LMI E2E Tests
+
+on:
+  workflow_dispatch:
+  pull_request:
+    types: [opened, synchronize, reopened]
+    paths:
+      - 'lmi-tests/**'
+      - 'sdk/**'
+      - 'pom.xml'
+      - '.github/workflows/lmi-e2e-tests.yml'
+
+permissions:
+  contents: read
+
+jobs:
+  local:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+      - uses: actions/checkout@v7
+      - uses: actions/setup-java@v6
+        with:
+          distribution: corretto
+          # Enforce the SDK's minimum JDK API compatibility for fixture-only changes.
+          java-version: '17'
+          cache: maven
+      - name: Build fixtures and validate evidence assertions
+        run: |
+          mvn -B -q -pl lmi-tests -am package
+          mvn -B -q -pl lmi-tests -am spotless:check
+          python3 -m unittest discover -s lmi-tests/tests -v
+
+  cloud:
+    if: github.event_name == 'workflow_dispatch'
+    needs: local
+    runs-on: ubuntu-latest
+    timeout-minutes: 95
+    permissions:
+      contents: read
+      id-token: write
+    # Persistent functions are shared; serialize only deployment and cloud testing.
+    concurrency:
+      group: java-lmi-e2e
+      cancel-in-progress: false
+    env:
+      LMI_STACK_NAME: java-lmi-e2e
+      CAPACITY_PROVIDER_ARN: ${{ secrets.CAPACITY_PROVIDER_ARN }}
+      TEST_LAMBDA_EXECUTION_ROLE_ARN: ${{ secrets.TEST_LAMBDA_EXECUTION_ROLE_ARN }}
+    steps:
+      - uses: actions/checkout@v7
+      - name: Validate configuration
+        shell: bash
+        run: |
+          set -euo pipefail
+          test -n "$CAPACITY_PROVIDER_ARN"
+          test -n "$TEST_LAMBDA_EXECUTION_ROLE_ARN"
+          region=$(cut -d: -f4 <<< "$CAPACITY_PROVIDER_ARN")
+          echo "AWS_REGION=$region" >> "$GITHUB_ENV"
+      - uses: actions/setup-java@v6
+        with:
+          distribution: corretto
+          java-version: '25'
+          cache: maven
+      - name: Build commit under test
+        timeout-minutes: 10
+        run: mvn -B -q -pl lmi-tests -am package -DskipTests
+      - name: Configure AWS credentials for deployment
+        uses: aws-actions/configure-aws-credentials@e1253824e5c10ff9df46874f81ed3ec929e19cfd # v6.3.0
+        with:
+          role-to-assume: ${{ secrets.TEST_ROLE_ARN }}
+          role-session-name: java-lmi-e2e-deploy
+          aws-region: ${{ env.AWS_REGION }}
+          allowed-account-ids: ${{ secrets.TEST_ACCOUNT_ID }}
+      - name: Create or update persistent LMI functions
+        timeout-minutes: 35
+        run: python3 lmi-tests/cloud_suite.py deploy --stack-name "$LMI_STACK_NAME" --run-id "${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
+      - name: Refresh AWS credentials for assertions
+        uses: aws-actions/configure-aws-credentials@e1253824e5c10ff9df46874f81ed3ec929e19cfd # v6.3.0
+        with:
+          role-to-assume: ${{ secrets.TEST_ROLE_ARN }}
+          role-session-name: java-lmi-e2e-test
+          aws-region: ${{ env.AWS_REGION }}
+          allowed-account-ids: ${{ secrets.TEST_ACCOUNT_ID }}
+      - name: LMI cloud regression assertions (expected red until issue 726 is fixed)
+        timeout-minutes: 30
+        run: python3 -u lmi-tests/cloud_suite.py test --cloud-enabled
+      - name: Refresh AWS credentials for evidence collection
+        if: always() && hashFiles('lmi-tests/artifacts/manifest.json') != ''
+        uses: aws-actions/configure-aws-credentials@e1253824e5c10ff9df46874f81ed3ec929e19cfd # v6.3.0
+        with:
+          role-to-assume: ${{ secrets.TEST_ROLE_ARN }}
+          role-session-name: java-lmi-e2e-collect
+          aws-region: ${{ env.AWS_REGION }}
+          allowed-account-ids: ${{ secrets.TEST_ACCOUNT_ID }}
+      - name: Collect histories and lifecycle evidence
+        if: always()
+        timeout-minutes: 5
+        run: |
+          if test -f lmi-tests/artifacts/manifest.json; then
+            python3 lmi-tests/cloud_suite.py collect
+          fi
+      - name: Publish failure evidence
+        if: always()
+        uses: actions/upload-artifact@v7
+        with:
+          name: lmi-e2e-${{ github.run_id }}-${{ github.run_attempt }}
+          retention-days: 7
+          path: |
+            lmi-tests/artifacts/
+            lmi-tests/target/surefire-reports/
+            sdk/target/surefire-reports/
+      - name: Summarize outcomes
+        if: always()
+        run: |
+          python3 - <<'PY'
+          import os, pathlib, xml.etree.ElementTree as ET
+          p = pathlib.Path('lmi-tests/artifacts/junit.xml')
+          with open(os.environ['GITHUB_STEP_SUMMARY'], 'a') as out:
+              out.write('LMI tests assert the fixed behavior in #726; failures are not expected-success results.\n\n')
+              if p.exists():
+                  for case in ET.parse(p).getroot().findall('testcase'):
+                      failures = list(case)
+                      state = 'PASS' if not failures else failures[0].get('type', 'FAIL')
+                      out.write(f"- {case.get('name')}: {state}\n")
+              else:
+                  out.write('Cloud assertions did not run. Inspect setup errors in the artifacts.\n')
+          PY

AGENTS.md

@@ -335,6 +335,7 @@ Run `mvn spotless:apply` after Java changes. Then run the narrowest relevant tes
 - [Error Handling](docs/advanced/error-handling.md)
 - [Logging](docs/advanced/logging.md)
 - [Migration from 1.x to 2.x](docs/migration-1.x-to-2.x.md)
+- [Migration from 2.x to 3.x](docs/migration-2.x-to-3.x.md)
 
 ### Official AWS SDKs
 

README.md

@@ -119,6 +119,7 @@ See [Deploy Lambda durable functions with Infrastructure as Code](https://docs.a
 - [<u>Error Handling</u>](docs/advanced/error-handling.md) - SDK exceptions for handling failures
 - [<u>Logging</u>](docs/advanced/logging.md) - How to use DurableLogger
 - [<u>Migrating from 1.x to 2.x</u>](docs/migration-1.x-to-2.x.md) - Upgrade guide for breaking changes since `v1.2.1`
+- [<u>Migrating from 2.x to 3.x</u>](docs/migration-2.x-to-3.x.md) - Upgrade guide for the factory-only, per-invocation plugin contract
 - [<u>Release Process</u>](RELEASE.md) - Prepare and publish Maven releases
 - [<u>Testing</u>](docs/advanced/testing.md) - Utilities for local development and cloud-based integration testing
 

conformance-tests-otel/README.md

@@ -35,13 +35,11 @@ SDK spans. The shared validator determines the canonical trace from SDK spans
 and checks the actual user-span parents.
 
 Root handler probes preserve a valid same-trace ambient Lambda parent. Otherwise
-the plugin activates its canonical Invocation/Workflow context on the handler
-thread and restores it in the same-thread end hook. The plugin requires a core with the
-same-thread invocation-hook lifecycle capability; unsupported cores are rejected
-during configuration.
-Nested callbacks retain their named context or attempt parents. Execution-view
-context placeholders can be valid without recording, so probes do not require
-`isRecording()`.
+the per-invocation plugin activates its canonical Invocation/Workflow context on the handler
+thread through its explicit handler scope and restores it on that same thread before End.
+Use the matching 3.x core and plugin factory API. Nested callbacks retain their named context
+or attempt parents. Execution-view context placeholders can be valid without recording, so
+probes do not require `isRecording()`.
 
 Case 23 uses the public step attempt number and checkpointed condition state;
 it has no mutable warm-container markers. Its helper and virtual child run

conformance-tests-otel/src/main/java/software/amazon/lambda/durable/conformance/otel/OtelConformanceHandler.java

@@ -9,7 +9,7 @@
 import software.amazon.lambda.durable.TypeToken;
 import software.amazon.lambda.durable.otel.ExecutionOtelPlugin;
 import software.amazon.lambda.durable.otel.InvocationOtelPlugin;
-import software.amazon.lambda.durable.plugin.DurableExecutionPlugin;
+import software.amazon.lambda.durable.plugin.DurableExecutionPluginFactory;
 
 /**
  * Shared base for the OTel conformance suite's handlers. Ported from the otel-invocation/otel-execution examples in
@@ -26,13 +26,13 @@ protected OtelConformanceHandler() {
 
     @Override
     protected final DurableConfig createConfiguration() {
-        return DurableConfig.builder().withPlugins(createPlugin()).build();
+        return DurableConfig.builder().withPlugins(createPluginFactory()).build();
     }
 
-    private DurableExecutionPlugin createPlugin() {
+    private DurableExecutionPluginFactory createPluginFactory() {
         return "execution".equals(System.getenv("OTEL_PLUGIN_MODE"))
-                ? new ExecutionOtelPlugin()
-                : new InvocationOtelPlugin();
+                ? ExecutionOtelPlugin.factory()
+                : InvocationOtelPlugin.factory();
     }
 
     protected final void requireScenario(Map<String, Object> event, String expected) {

conformance-tests/pom.xml

@@ -7,7 +7,7 @@
     <parent>
         <groupId>software.amazon.lambda.durable</groupId>
         <artifactId>aws-durable-execution-sdk-java-parent</artifactId>
-        <version>2.3.1-SNAPSHOT</version>
+        <version>3.0.0-SNAPSHOT</version>
     </parent>
 
     <artifactId>aws-durable-execution-sdk-java-conformance-tests</artifactId>

conformance-tests/src/main/java/plugin/PluginAttemptHooksRetry.java

@@ -22,7 +22,7 @@ public class PluginAttemptHooksRetry extends DurableHandler<Object, String> {
     @Override
     protected DurableConfig createConfiguration() {
         return DurableConfig.builder()
-                .withPlugins(new ConformanceLoggingPlugin("CONFPLUGIN"))
+                .withPlugins(info -> new ConformanceLoggingPlugin("CONFPLUGIN"))
                 .build();
     }
 

conformance-tests/src/main/java/plugin/PluginErrorIsolation.java

@@ -18,7 +18,7 @@ public class PluginErrorIsolation extends DurableHandler<String, String> {
     @Override
     protected DurableConfig createConfiguration() {
         return DurableConfig.builder()
-                .withPlugins(new FaultyConformancePlugin())
+                .withPlugins(info -> new FaultyConformancePlugin())
                 .build();
     }
 

conformance-tests/src/main/java/plugin/PluginExternalUpdateOnInvoke.java

@@ -23,7 +23,9 @@ public class PluginExternalUpdateOnInvoke extends DurableHandler<Object, String>
 
     @Override
     protected DurableConfig createConfiguration() {
-        return DurableConfig.builder().withPlugins(new UpdatedOnInvokePlugin()).build();
+        return DurableConfig.builder()
+                .withPlugins(info -> new UpdatedOnInvokePlugin())
+                .build();
     }
 
     @Override