xds: Implement Context propagation in ExternalProcessorClientInterceptor

Ensure request-scoped io.grpc.Context is correctly captured and propagated
across asynchronous thread boundaries during the external processor client-side
interceptor lifecycle.

Previously, asynchronous boundaries—including delayed application call
activation and downstream listener callbacks—ran on transport executor
threads without the request-scoped context, leading to lost telemetry, tracing,
and deadline propagation.

This change:
- Captures the active context as 'callContext' inside DataPlaneClientCall.start(),
  representing the exact point when the application starts the RPC.
- Wraps delayed call activation execution in 'callContext.run(...)' to ensure
  that downstream transport activation runs under the captured context.
- Refactors DataPlaneListener callback invocations to wrap downstream delegate
  calls (onHeaders, onMessage, onClose, onReady) in 'callContext.run(...)'.
- Relies on implicit context propagation during stub.process() stream
  initiation on the calling application thread.
- Adds Category 19 unit tests verifying context propagation to startCall,
  listener callbacks, and the outbound ext_proc stream stub.
- Fixes checkstyle style compliance in main and test source classes.

Author: kannanjgithub

xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java

@@ -47,6 +47,7 @@
 import io.grpc.Channel;
 import io.grpc.ClientCall;
 import io.grpc.ClientInterceptor;
+import io.grpc.Context;
 import io.grpc.Deadline;
 import io.grpc.Detachable;
 import io.grpc.DoubleHistogramMetricInstrument;
@@ -434,6 +435,7 @@ private enum EventType {
     private final MetricRecorder metricsRecorder;
     private final String target;
     private final String backendService;
+    private volatile Context callContext = Context.ROOT;
 
     private long clientHeadersStartNanos;
     private long clientHalfCloseStartNanos;
@@ -526,7 +528,7 @@ private void activateCall() {
       }
       Runnable toRun = delayedCall.setCall(rawCall);
       if (toRun != null) {
-        toRun.run();
+        callContext.run(toRun);
       }
       drainPendingRequests();
       onReadyNotify();
@@ -611,6 +613,7 @@ private void applyHeaderMutations(Metadata metadata,
 
     @Override
     public void start(Listener<InputStream> responseListener, Metadata headers) {
+      this.callContext = Context.current();
       clientHeadersStartNanos = System.nanoTime();
       this.requestHeaders = headers;
       this.wrappedListener = new DataPlaneListener(responseListener, rawCall, this);
@@ -1204,7 +1207,7 @@ public void onHeaders(Metadata headers) {
     @Override
     public void onMessage(InputStream message) {
       if (dataPlaneClientCall.passThroughMode.get()) {
-        delegate().onMessage(message);
+        dataPlaneClientCall.callContext.run(() -> delegate().onMessage(message));
         return;
       }
 
@@ -1216,7 +1219,7 @@ public void onMessage(InputStream message) {
       if (dataPlaneClientCall.extProcStreamState.get().isCompleted()
           || dataPlaneClientCall.currentProcessingMode.getResponseBodyMode()
               != ProcessingMode.BodySendMode.GRPC) {
-        delegate().onMessage(message);
+        dataPlaneClientCall.callContext.run(() -> delegate().onMessage(message));
         return;
       }
 
@@ -1225,7 +1228,8 @@ public void onMessage(InputStream message) {
         sendResponseBodyToExtProc(bodyByteString, false);
 
         if (dataPlaneClientCall.config.getObservabilityMode()) {
-          delegate().onMessage(new InboundZeroCopyInputStream(bodyByteString));
+          dataPlaneClientCall.callContext.run(
+              () -> delegate().onMessage(new InboundZeroCopyInputStream(bodyByteString)));
         }
       } catch (IOException e) {
         rawCall.cancel("Failed to read server response", e);
@@ -1275,7 +1279,7 @@ public void onClose(Status status, Metadata trailers) {
     }
 
     void onReadyNotify() {
-      delegate().onReady();
+      dataPlaneClientCall.callContext.run(() -> delegate().onReady());
     }
 
     void proceedWithHeaders() {
@@ -1299,7 +1303,7 @@ private void proceedWithHeaders(Metadata headers) {
         dataPlaneClientCall.recordDuration(serverHeadersDuration, durationNanos);
         dataPlaneClientCall.serverHeadersStartNanos = 0;
       }
-      delegate().onHeaders(headers);
+      dataPlaneClientCall.callContext.run(() -> delegate().onHeaders(headers));
     }
 
     void proceedWithClose() {
@@ -1318,11 +1322,12 @@ private void proceedWithClose(Status status, Metadata trailers) {
         dataPlaneClientCall.recordDuration(serverTrailersDuration, durationNanos);
         dataPlaneClientCall.serverTrailersStartNanos = 0;
       }
-      delegate().onClose(status, trailers);
+      dataPlaneClientCall.callContext.run(() -> delegate().onClose(status, trailers));
     }
 
     void onExternalBody(ByteString body) {
-      delegate().onMessage(new InboundZeroCopyInputStream(body));
+      dataPlaneClientCall.callContext.run(
+          () -> delegate().onMessage(new InboundZeroCopyInputStream(body)));
     }
 
     void unblockAfterStreamComplete() {