fix(serviceevents): gate incident correlation on SAMPLED + per-collector fault isolation (#1416)

## Description

Refines the ServiceEvents span processor and incident-snapshot pipeline
for the Java distro. Builds on the initial ServiceEvents feature
(#1386). Aligns incident-route derivation with the Python/JS distros.

Highlights:
- **Incident trace correlation gated on SAMPLED** — an incident snapshot
carries `trace_id`/`span_id` only when the request's trace was actually
sampled; also removes dead lite-mode drainer code.
- **Per-collector fault isolation** — `recordEndpointRequest` and
`recordPotentialIncident` are wrapped in separate try/catch blocks in
`processRequestSpan`, so a crash in one collector no longer skips the
other.
- **Incident route derived from the ingress operation** — aligns route
derivation with the Python/JS distros.

## Testing

- `./gradlew build` — BUILD SUCCESSFUL (compile + unit tests + spotless
+ contract-test assembly).
- Validated end-to-end against live CloudWatch (ServiceEvents contract
tests + validators).

## Type of change

- [x] Bug fix (non-breaking change which fixes an issue)
- [x] Refactor / code hardening

## Checklist

- [x] Unit tests added/updated and passing
- [x] `CHANGELOG.md` updated

Author: vastin

CHANGELOG.md

@@ -13,6 +13,7 @@ If your change does not need a CHANGELOG entry, add the "skip changelog" label t
 
 ## Unreleased
 
+- fix(serviceevents): gate incident trace correlation on the SAMPLED flag
 - fix: remove EOL AWS SDK v1 dependency for ARN parsing
   ([#1401](https://github.com/aws-observability/aws-otel-java-instrumentation/pull/1401))
 

appsignals-tests/contract-tests/src/test/java/software/amazon/opentelemetry/appsignals/test/serviceevents/base/ServiceEventsContractTestBase.java

@@ -519,19 +519,24 @@ protected JsonNode waitForEndpointRecord(String method, String route) throws Exc
   /**
    * Assert that an IncidentSnapshot has valid telemetry correlation fields.
    *
-   * <p>Validates that request_id is present (always generated) and optionally checks trace/span
-   * IDs.
+   * <p>Trace correlation is sampling-conditional: the SpanProcessor only attaches trace_id/span_id
+   * when the request's trace was sampled (otherwise the link would point at a trace the backend
+   * never received). This suite forces OTEL_TRACES_SAMPLER=always_on, so every request IS sampled
+   * and both ids are always populated and non-empty. Under reduced sampling an unsampled request
+   * would still emit a complete IncidentSnapshot, but with the correlation ids omitted. (Matches
+   * the Node + Python serviceevents contract suites.)
    */
   protected void assertTelemetryCorrelation(JsonNode record) {
     JsonNode correlation = record.path("telemetry_correlation");
     assertThat(correlation.isMissingNode()).isFalse();
-    // telemetry_correlation should have at least one identifier present: the synchronous
-    // IncidentSnapshot path provides request_id, and trace_id/span_id when the request is sampled.
-    boolean hasAnyCorrelation =
-        correlation.has("request_id") || correlation.has("trace_id") || correlation.has("span_id");
-    assertThat(hasAnyCorrelation)
-        .as("telemetry_correlation should have at least one of: request_id, trace_id, span_id")
+    assertThat(correlation.has("trace_id"))
+        .as("telemetry_correlation should carry trace_id (request is sampled under always_on)")
         .isTrue();
+    assertThat(correlation.path("trace_id").asText()).isNotEmpty();
+    assertThat(correlation.has("span_id"))
+        .as("telemetry_correlation should carry span_id (request is sampled under always_on)")
+        .isTrue();
+    assertThat(correlation.path("span_id").asText()).isNotEmpty();
   }
 
   /**

appsignals-tests/contract-tests/src/test/java/software/amazon/opentelemetry/appsignals/test/serviceevents/springmvc/ServiceEventsSpringMvcTest.java

@@ -567,21 +567,16 @@ void testIncidentSnapshotRequestContext() {
   }
 
   /**
-   * Verifies telemetry_correlation field is present on incident snapshots. The correlation object
-   * may contain trace_id/span_id (from SpanProcessor) or request_id (from direct collector), or may
-   * be empty if correlation data was not available at capture time.
+   * Verifies telemetry_correlation on incident snapshots carries trace_id/span_id. Correlation is
+   * sampling-conditional; this suite forces always_on so both ids are always populated. See {@link
+   * ServiceEventsContractTestBase#assertTelemetryCorrelation} for the full rationale.
    */
   @Test
   @Order(1)
   void testIncidentSnapshotTelemetryCorrelation() {
     assumeFileSignalsPresent(incidentRecords, "IncidentSnapshot");
     assertThat(incidentRecords).isNotEmpty();
-    JsonNode correlation = incidentRecords.get(0).path("telemetry_correlation");
-    assertThat(correlation.isMissingNode()).isFalse();
-    // The telemetry_correlation object should exist as a JSON object.
-    // It may contain trace_id, span_id, request_id, or be empty if the incident
-    // was recorded without an active OTel span context.
-    assertThat(correlation.isObject()).isTrue();
+    assertTelemetryCorrelation(incidentRecords.get(0));
   }
 
   @Test

instrumentation/serviceevents/src/main/java/software/amazon/opentelemetry/javaagent/instrumentation/serviceevents/ServiceEventsInstrumentation.java

@@ -326,9 +326,8 @@ public synchronized void initialize() {
         }
       }
 
-      // Build the shared IncidentSnapshotRecordBuilder. Used by:
-      //   - bytecode mode: synchronous IncidentSnapshotEmitter
-      //   - lite mode (no bytecode): asynchronous LiteIncidentDrainer
+      // Build the shared IncidentSnapshotRecordBuilder for the synchronous IncidentSnapshotEmitter
+      // (used in both bytecode and lite mode; lite mode simply omits the call_path).
       software.amazon.opentelemetry.javaagent.instrumentation.serviceevents.utils
               .IncidentSnapshotRecordBuilder
           incidentRecordBuilder =
@@ -357,8 +356,11 @@ public synchronized void initialize() {
                   + ", maxSameError="
                   + config.getIncidentSnapshotMaxSameError());
 
-      // Always install the direct-emit IncidentSnapshot bridge so incidents emit synchronously.
-      // Call path is only captured when bytecode instrumentation is enabled.
+      // Install the direct-emit IncidentSnapshot bridge so incidents emit synchronously in both
+      // modes. Call path is only captured when bytecode instrumentation is enabled; in lite mode
+      // the
+      // snapshot carries exception_info.stack_trace but no call_path (matches Python lite-mode
+      // shape).
       software.amazon.opentelemetry.javaagent.instrumentation.serviceevents.exporter
               .IncidentSnapshotEmitter
           incidentEmitter =
@@ -368,29 +370,6 @@ public synchronized void initialize() {
           .setIncidentSnapshotEmitterBridge(incidentEmitter);
       logger().info("Installed IncidentSnapshotEmitter: incidents emit synchronously");
 
-      if (!config.isBytecodeEnabled()) {
-        // Lite mode: no bytecode advice. Install the in-memory drainer as the
-        // MetadataWriterBridge so DataStore.recordPotentialIncident dispatches to its
-        // queue, and ticks records out via the existing OTLP path on the configured
-        // flush interval. Snapshots carry exception_info.stack_trace but no call_path
-        // (matches Python lite-mode shape).
-        software.amazon.opentelemetry.javaagent.instrumentation.serviceevents.collectors
-                .LiteIncidentDrainer
-            liteDrainer =
-                new software.amazon.opentelemetry.javaagent.instrumentation.serviceevents.collectors
-                    .LiteIncidentDrainer(10_000, incidentRecordBuilder, otlpEmitter);
-        software.amazon.opentelemetry.serviceevents.ServiceEventsDataStore.setMetadataWriterBridge(
-            liteDrainer);
-        collectors.add(liteDrainer);
-        liteDrainer.start();
-        logger()
-            .info(
-                "Installed LiteIncidentDrainer (lite mode: no bytecode). "
-                    + "Incidents queue in-memory and emit every "
-                    + 10_000
-                    + "ms.");
-      }
-
       initialized = true;
       logger()
           .info(

instrumentation/serviceevents/src/main/java/software/amazon/opentelemetry/javaagent/instrumentation/serviceevents/ServiceEventsSpanProcessor.java

@@ -78,8 +78,6 @@ private static Logger logger() {
       AttributeKey.stringKey("exception.stacktrace");
 
   // HTTP semantic convention attribute keys
-  private static final AttributeKey<String> HTTP_ROUTE = AttributeKey.stringKey("http.route");
-  private static final AttributeKey<String> URL_PATH = AttributeKey.stringKey("url.path");
   private static final AttributeKey<String> HTTP_REQUEST_METHOD =
       AttributeKey.stringKey("http.request.method");
   private static final AttributeKey<Long> HTTP_RESPONSE_STATUS_CODE =
@@ -144,10 +142,16 @@ public void onEnd(ReadableSpan span) {
   private void processRequestSpan(ReadableSpan span) {
     SpanData spanData = span.toSpanData();
 
-    // Extract trace context
+    // Extract trace context. Trace correlation is best-effort and sampling-conditional: under
+    // reduced sampling, AlwaysRecordSampler keeps this span recording (so App Signals metrics see
+    // every request) even though its trace was dropped before export — a valid but unsampled
+    // (RECORD_ONLY) span. Capturing its ids would emit a correlation link to a trace the backend
+    // never received, so gate on isSampled() in addition to isValid(). An unsampled request still
+    // emits a complete (self-contained) IncidentSnapshot, just without trace_id/span_id.
     SpanContext spanContext = spanData.getSpanContext();
-    String traceId = spanContext.isValid() ? spanContext.getTraceId() : null;
-    String spanId = spanContext.isValid() ? spanContext.getSpanId() : null;
+    boolean sampled = spanContext.isValid() && spanContext.isSampled();
+    String traceId = sampled ? spanContext.getTraceId() : null;
+    String spanId = sampled ? spanContext.getSpanId() : null;
 
     // Extract exception from span events
     String exceptionType = null;
@@ -163,19 +167,17 @@ private void processRequestSpan(ReadableSpan span) {
       }
     }
 
-    // Record endpoint/incident/span data directly from SpanData. This gives us the correct
-    // status code (set by the container after the servlet exits) and exception data in one
-    // place.
-    String route = spanData.getAttributes().get(HTTP_ROUTE);
-    if (route == null || route.isEmpty()) {
-      route = spanData.getAttributes().get(URL_PATH);
-    }
-    if (route == null || route.isEmpty()) {
+    String method = spanData.getAttributes().get(HTTP_REQUEST_METHOD);
+    if (method == null || method.isEmpty()) {
       return;
     }
 
-    String method = spanData.getAttributes().get(HTTP_REQUEST_METHOD);
-    if (method == null || method.isEmpty()) {
+    // Derive the operation via the shared App Signals path (span-name primary, first-path-segment
+    // fallback) — consistent with Python/JS and with what App Signals reports. Then back the
+    // route out of the operation so the collector rebuilds the identical operation string.
+    String operation = AwsSpanProcessingUtil.getIngressOperation(spanData);
+    String route = routeFromOperation(operation, method);
+    if (route == null) {
       return;
     }
 
@@ -199,8 +201,6 @@ private void processRequestSpan(ReadableSpan span) {
       threadName = Thread.currentThread().getName();
     }
 
-    String operation = AwsSpanProcessingUtil.getIngressOperation(spanData);
-
     // Derive simple error type from FQCN (e.g. "java.lang.RuntimeException" ->
     // "RuntimeException")
     String errorType = null;
@@ -230,8 +230,13 @@ private void processRequestSpan(ReadableSpan span) {
     }
 
     // 1. Record endpoint request
-    ServiceEventsDataStore.recordEndpointRequest(
-        operation, route, method, statusCode, durationNs, errorType, errorFunctionId, operation);
+    try {
+      ServiceEventsDataStore.recordEndpointRequest(
+          operation, route, method, statusCode, durationNs, errorType, errorFunctionId, operation);
+    } catch (Exception e) {
+      logger()
+          .log(Level.WARNING, "[SERVICE_EVENTS-SPAN-PROCESSOR] recordEndpointRequest failed", e);
+    }
 
     // Set currentOperation so recordPotentialIncident can attach exemplars
     // to the EndpointAggregation. The thread-local may have been cleared by
@@ -241,22 +246,27 @@ private void processRequestSpan(ReadableSpan span) {
     ServiceEventsDataStore.setCurrentOperation(operation);
 
     // 2. Record potential incident
-    ServiceEventsDataStore.recordPotentialIncident(
-        route,
-        method,
-        statusCode,
-        durationMs,
-        exceptionType,
-        exceptionMessage,
-        stackTrace,
-        null, // headers
-        null, // queryParams
-        threadName,
-        startTimeNs,
-        endTimeNs,
-        traceId,
-        spanId,
-        operation);
+    try {
+      ServiceEventsDataStore.recordPotentialIncident(
+          route,
+          method,
+          statusCode,
+          durationMs,
+          exceptionType,
+          exceptionMessage,
+          stackTrace,
+          null, // headers
+          null, // queryParams
+          threadName,
+          startTimeNs,
+          endTimeNs,
+          traceId,
+          spanId,
+          operation);
+    } catch (Exception e) {
+      logger()
+          .log(Level.WARNING, "[SERVICE_EVENTS-SPAN-PROCESSOR] recordPotentialIncident failed", e);
+    }
   }
 
   /**
@@ -270,6 +280,42 @@ private static boolean isLocalRoot(ReadableSpan span) {
     return !parentContext.isValid() || parentContext.isRemote();
   }
 
+  /**
+   * Back the route out of the App Signals operation so the collector rebuilds the identical {@code
+   * method + " " + route} operation string.
+   *
+   * <p>Handles the three shapes {@code getIngressOperation} returns:
+   *
+   * <ul>
+   *   <li>{@code "METHOD /route"} — common case. Strip the method prefix.
+   *   <li>{@code "/route"} — bare path (no method prefix). Use verbatim.
+   *   <li>{@code InternalOperation / UnknownOperation / bare method / lambda} — no resolvable
+   *       route. Return null so the caller skips.
+   * </ul>
+   *
+   * <p>Matches Python's {@code _route_from_operation} and JS's {@code routeFromOperation}.
+   */
+  static String routeFromOperation(String operation, String method) {
+    if (operation == null || operation.isEmpty()) {
+      return null;
+    }
+    if ("InternalOperation".equals(operation) || "UnknownOperation".equals(operation)) {
+      return null;
+    }
+    if (operation.equals(method)) {
+      return null;
+    }
+    String prefix = method + " ";
+    if (operation.startsWith(prefix)) {
+      String route = operation.substring(prefix.length());
+      return route.isEmpty() ? null : route;
+    }
+    if (operation.startsWith("/")) {
+      return operation;
+    }
+    return null;
+  }
+
   /**
    * Extract "com.example.MyClass.myMethod" from the first "at" line of a Java stack trace.
    *