@@ -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);
             }
           }
         }
 
@@ -90,9 +90,11 @@
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -21129,6 +21131,164 @@ public void onClose(Status status, Metadata trailers) {
     channelManager.close();
   }
 
+  @Test
+  public void whenCallCancelledConcurrently_underlyingCallCancelledExactlyOnceAndNoExtProcLeak()
+      throws Exception {
+    String uniqueExtProcServerName = InProcessServerBuilder.generateName();
+    String uniqueDataPlaneServerName = InProcessServerBuilder.generateName();
+
+    ExternalProcessor proto = createBaseProto(uniqueExtProcServerName)
+        .setProcessingMode(ProcessingMode.newBuilder()
+            .setRequestHeaderMode(ProcessingMode.HeaderSendMode.SEND)
+            .setResponseHeaderMode(ProcessingMode.HeaderSendMode.SEND)
+            .build())
+        .build();
+    ExternalProcessorFilterConfig filterConfig =
+        provider.parseFilterConfig(Any.pack(proto), filterContext).config;
+
+    final AtomicInteger extProcErrorCount = new AtomicInteger();
+    final AtomicInteger extProcRequestsAfterError = new AtomicInteger();
+    final AtomicBoolean extProcHadError = new AtomicBoolean(false);
+
+    ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl =
+        new ExternalProcessorGrpc.ExternalProcessorImplBase() {
+          @Override
+          public StreamObserver<ProcessingRequest> process(
+              StreamObserver<ProcessingResponse> responseObserver) {
+            return new StreamObserver<ProcessingRequest>() {
+              @Override
+              public void onNext(ProcessingRequest request) {
+                if (extProcHadError.get()) {
+                  extProcRequestsAfterError.incrementAndGet();
+                }
+                if (request.hasRequestHeaders()) {
+                  responseObserver.onNext(ProcessingResponse.newBuilder()
+                      .setRequestHeaders(HeadersResponse.newBuilder().build())
+                      .build());
+                }
+              }
+
+              @Override
+              public void onError(Throwable t) {
+                extProcHadError.set(true);
+                extProcErrorCount.incrementAndGet();
+              }
+
+              @Override
+              public void onCompleted() {
+                responseObserver.onCompleted();
+              }
+            };
+          }
+        };
+
+    grpcCleanup.register(InProcessServerBuilder.forName(uniqueExtProcServerName)
+        .addService(extProcImpl)
+        .directExecutor()
+        .build().start());
+
+    CachedChannelManager channelManager = new CachedChannelManager(config -> {
+      return grpcCleanup.register(
+          InProcessChannelBuilder.forName(uniqueExtProcServerName).directExecutor().build());
+    });
+
+    ExternalProcessorClientInterceptor interceptor = new ExternalProcessorClientInterceptor(
+        filterConfig, channelManager, scheduler, FAKE_CONTEXT);
+
+    final CountDownLatch dataPlaneServerStartedLatch = new CountDownLatch(1);
+    final CountDownLatch releaseServerResponseLatch = new CountDownLatch(1);
+
+    MutableHandlerRegistry dataPlaneRegistry = new MutableHandlerRegistry();
+    dataPlaneRegistry.addService(ServerServiceDefinition.builder("test.TestService")
+        .addMethod(METHOD_SAY_HELLO, ServerCalls.asyncUnaryCall((request, responseObserver) -> {
+          dataPlaneServerStartedLatch.countDown();
+          try {
+            releaseServerResponseLatch.await(5, TimeUnit.SECONDS);
+          } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+          }
+          responseObserver.onNext("Hello " + request);
+          responseObserver.onCompleted();
+        }))
+        .build());
+
+    grpcCleanup.register(InProcessServerBuilder.forName(uniqueDataPlaneServerName)
+        .fallbackHandlerRegistry(dataPlaneRegistry)
+        .directExecutor()
+        .build().start());
+
+    final AtomicInteger rawCallCancelCount = new AtomicInteger();
+    ClientInterceptor countingInterceptor = new ClientInterceptor() {
+      @Override
+      public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+          MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
+        return new SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
+          @Override
+          public void cancel(String message, Throwable cause) {
+            rawCallCancelCount.incrementAndGet();
+            super.cancel(message, cause);
+          }
+        };
+      }
+    };
+
+    ManagedChannel dataPlaneChannel = grpcCleanup.register(
+        InProcessChannelBuilder.forName(uniqueDataPlaneServerName)
+            .intercept(countingInterceptor)
+            .directExecutor()
+            .build());
+
+    ClientCall<String, String> proxyCall = interceptCall(
+        interceptor,
+        METHOD_SAY_HELLO,
+        DEFAULT_CALL_OPTIONS.withExecutor(MoreExecutors.directExecutor()),
+        dataPlaneChannel);
+
+    proxyCall.start(new ClientCall.Listener<String>() {}, new Metadata());
+
+    proxyCall.request(1);
+    proxyCall.sendMessage("ping");
+    proxyCall.halfClose();
+
+    // Wait until the data plane call has started and is active
+    assertThat(dataPlaneServerStartedLatch.await(5, TimeUnit.SECONDS)).isTrue();
+
+    // Concurrently release server response and trigger multiple concurrent cancellations
+    int numCancelThreads = 8;
+    ExecutorService executor = Executors.newFixedThreadPool(numCancelThreads + 1);
+    CyclicBarrier barrier = new CyclicBarrier(numCancelThreads + 1);
+    List<Future<?>> futures = new ArrayList<>();
+
+    futures.add(executor.submit(() -> {
+      barrier.await();
+      releaseServerResponseLatch.countDown();
+      return null;
+    }));
+
+    for (int i = 0; i < numCancelThreads; i++) {
+      final int threadId = i;
+      futures.add(executor.submit(() -> {
+        barrier.await();
+        proxyCall.cancel("Cancel from thread " + threadId, null);
+        return null;
+      }));
+    }
+
+    for (Future<?> f : futures) {
+      f.get(5, TimeUnit.SECONDS);
+    }
+
+    // Underlying rawCall was cancelled exactly once despite multiple concurrent cancellations
+    assertThat(rawCallCancelCount.get()).isEqualTo(1);
+    // Ext-proc received onError at most once
+    assertThat(extProcErrorCount.get()).isAtMost(1);
+    // Ext-proc never received any requests after being cancelled
+    assertThat(extProcRequestsAfterError.get()).isEqualTo(0);
+
+    channelManager.close();
+    shutdownAndAwaitTermination(executor);
+  }
+
   private static List<ProcessingRequest> filterClientRequests(List<ProcessingRequest> requests) {
     List<ProcessingRequest> clientRequests = new ArrayList<>();
     for (ProcessingRequest r : requests) {