diff --git a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java
index 04830b29fa7..f9352d51826 100644
--- a/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java
+++ b/xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java
@@ -236,8 +236,18 @@ public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
     MethodDescriptor<InputStream, InputStream> rawMethod =
         (MethodDescriptor<InputStream, InputStream>) (MethodDescriptor<?, ?>) method;
     ClientCall<InputStream, InputStream> rawCall =
-        (ClientCall<InputStream, InputStream>) (ClientCall<?, ?>)
-            next.newCall(method, callOptions);
+        new SimpleForwardingClientCall<InputStream, InputStream>(
+            (ClientCall<InputStream, InputStream>) (ClientCall<?, ?>)
+                next.newCall(method, callOptions)) {
+          private final AtomicBoolean cancelled = new AtomicBoolean(false);
+
+          @Override
+          public void cancel(@Nullable String message, @Nullable Throwable cause) {
+            if (cancelled.compareAndSet(false, true)) {
+              super.cancel(message, cause);
+            }
+          }
+        };
 
     // Create a local subclass instance to buffer outbound actions
     DataPlaneDelayedCall<InputStream, InputStream> delayedCall =
@@ -362,7 +372,6 @@ private static class DataPlaneClientCall
     final AtomicBoolean isProcessingTrailers = new AtomicBoolean(false);
     final AtomicBoolean pendingHalfClose = new AtomicBoolean(false);
     final AtomicBoolean bodyMessageSentToExtProc = new AtomicBoolean(false);
-    private final AtomicBoolean downstreamCancelled = new AtomicBoolean(false);
 
     protected DataPlaneClientCall(
         DataPlaneDelayedCall<InputStream, InputStream> delayedCall,
@@ -443,13 +452,14 @@ private boolean validateCompressionSupport(BodyResponse bodyResponse) {
               .withDescription("gRPC message compression not supported in ext_proc")
               .asRuntimeException();
           synchronized (streamLock) {
-            if (!extProcStreamState.get().isCompleted()
-                && extProcClientCallRequestObserver != null) {
-              extProcClientCallRequestObserver.onError(ex);
+            if (markExtProcStreamFailed(extProcStreamState)) {
+              if (extProcClientCallRequestObserver != null) {
+                extProcClientCallRequestObserver.onError(ex);
+                extProcClientCallRequestObserver = null;
+              }
             }
           }
           activateCall();
-          markExtProcStreamFailed(extProcStreamState);
           cancelDownstream("gRPC message compression not supported in ext_proc", ex);
           closeExtProcStream();
           return false;
@@ -463,7 +473,7 @@ 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);
+      this.wrappedListener = new DataPlaneListener(responseListener, this);
 
       // DelayedClientCall.start will buffer the listener and headers until setCall is called.
       super.start(wrappedListener, headers);
@@ -664,6 +674,9 @@ public void onError(Throwable t) {
         @Override
         public void onCompleted() {
           if (markExtProcStreamCompleted(extProcStreamState)) {
+            synchronized (streamLock) {
+              extProcClientCallRequestObserver = null;
+            }
             handleFailOpen(wrappedListener);
           }
         }
@@ -693,7 +706,7 @@ public void onCompleted() {
 
     private void sendToExtProc(ProcessingRequest request) {
       synchronized (streamLock) {
-        if (extProcStreamState.get().isCompleted()) {
+        if (extProcStreamState.get().isCompleted() || extProcClientCallRequestObserver == null) {
           return;
         }
         
@@ -830,6 +843,7 @@ private void closeExtProcStream() {
         if (markExtProcStreamCompleted(extProcStreamState)) {
           if (extProcClientCallRequestObserver != null) {
             extProcClientCallRequestObserver.onCompleted();
+            extProcClientCallRequestObserver = null;
           }
         }
       }
@@ -839,11 +853,7 @@ private void internalOnError(Throwable t) {
       if (markExtProcStreamFailed(extProcStreamState)) {
         synchronized (streamLock) {
           if (extProcClientCallRequestObserver != null) {
-            try {
-              extProcClientCallRequestObserver.onError(t);
-            } catch (Throwable ignored) {
-              // Ignore exceptions during cancel/onError propagation
-            }
+            extProcClientCallRequestObserver.onError(t);
             extProcClientCallRequestObserver = null;
           }
         }
@@ -999,7 +1009,7 @@ public void sendMessage(InputStream message) {
             ByteString copiedBody = ByteString.readFrom(message);
             pendingDrainingMessages.add(new KnownLengthInputStream(copiedBody));
           } catch (IOException e) {
-            rawCall.cancel("Failed to copy outbound message for buffering", e);
+            cancelDownstream("Failed to copy outbound message for buffering", e);
           }
           return;
         }
@@ -1029,7 +1039,7 @@ public void sendMessage(InputStream message) {
             }
           }
         } catch (IOException e) {
-          rawCall.cancel("Failed to serialize message for External Processor", e);
+          cancelDownstream("Failed to serialize message for External Processor", e);
         }
       }
     }
@@ -1131,21 +1141,22 @@ public void halfClose() {
       }
     }
 
-    private void cancelDownstream(@Nullable String message, @Nullable Throwable cause) {
-      if (downstreamCancelled.compareAndSet(false, true)) {
-        delayedCall.cancel(message, cause);
-      }
+    void cancelDownstream(@Nullable String message, @Nullable Throwable cause) {
+      delayedCall.cancel(message, cause);
     }
 
     @Override
     public void cancel(@Nullable String message, @Nullable Throwable cause) {
       synchronized (streamLock) {
-        if (!extProcStreamState.get().isCompleted() && extProcClientCallRequestObserver != null) {
-          extProcClientCallRequestObserver.onError(
-              Status.CANCELLED
-                  .withDescription(message)
-                  .withCause(cause)
-                  .asRuntimeException());
+        if (markExtProcStreamFailed(extProcStreamState)) {
+          if (extProcClientCallRequestObserver != null) {
+            extProcClientCallRequestObserver.onError(
+                Status.CANCELLED
+                    .withDescription(message)
+                    .withCause(cause)
+                    .asRuntimeException());
+            extProcClientCallRequestObserver = null;
+          }
         }
       }
       cancelDownstream(message, cause);
@@ -1318,7 +1329,7 @@ private void handleImmediateResponse(ImmediateResponse immediate, DataPlaneListe
         // If sent in response to any other event, it will cause the data plane RPC to
         // immediately fail with the specified status as if it were an out-of-band
         // cancellation.
-        rawCall.cancel(status.getDescription(), null);
+        cancelDownstream(status.getDescription(), null);
         listener.unblockAfterStreamComplete();
       }
       closeExtProcStream();
@@ -1442,7 +1453,6 @@ AtomicBoolean getIsProcessingTrailers() {
   }
 
   private static class DataPlaneListener extends SimpleForwardingClientCallListener<InputStream> {
-    private final ClientCall<?, ?> rawCall;
     private final DataPlaneClientCall dataPlaneClientCall;
     // Path 3: Upstream response bodies queued because upstream to sidestream window not available,
     // response headers not cleared by ext_proc or ext_proc stream draining
@@ -1457,10 +1467,8 @@ private static class DataPlaneListener extends SimpleForwardingClientCallListene
 
     protected DataPlaneListener(
         ClientCall.Listener<InputStream> delegate,
-        ClientCall<?, ?> rawCall,
         DataPlaneClientCall dataPlaneClientCall) {
       super(delegate);
-      this.rawCall = rawCall;
       this.dataPlaneClientCall = dataPlaneClientCall;
     }
 
@@ -1538,7 +1546,7 @@ public void onMessage(InputStream message) {
             ByteString copiedBody = ByteString.readFrom(message);
             savedMessages.add(new KnownLengthInputStream(copiedBody));
           } catch (IOException e) {
-            rawCall.cancel("Failed to copy inbound message for buffering", e);
+            dataPlaneClientCall.cancelDownstream("Failed to copy inbound message for buffering", e);
           }
           return;
         }
@@ -1574,7 +1582,7 @@ public void onMessage(InputStream message) {
             dataPlaneClientCall.drainPendingRequests();
           }
         } catch (IOException e) {
-          rawCall.cancel("Failed to read server response", e);
+          dataPlaneClientCall.cancelDownstream("Failed to read server response", e);
         }
       }
     }
@@ -1592,7 +1600,7 @@ void drainSavedMessages() {
               sendResponseBodyToExtProc(bodyByteString, false);
               dataPlaneClientCall.bodyMessageSentToExtProc.set(true);
             } catch (IOException e) {
-              rawCall.cancel("Failed to read buffered response body", e);
+              dataPlaneClientCall.cancelDownstream("Failed to read buffered response body", e);
             }
           }
         }