I have implemented heap-copy approach for both buffering queues in the interceptor:

Inbound Queue (savedMessages): In DataPlaneListener.onMessage, when a message is queued (because we are waiting for external processor header responses), it is now synchronously copied into a heap-allocated ByteString and queued as a KnownLengthInputStream. This prevents gRPC's deframer from recycling the underlying Netty direct buffers when onMessage returns, fully resolving the inbound Use-After-Free risk.
Outbound Queue (pendingDrainingMessages): In DataPlaneClientCall.sendMessage, when outbound messages are queued (because the ext-proc stream is draining or completing), we now synchronously read the stream into a heap-allocated ByteString and queue it as a KnownLengthInputStream before returning immediately to the application. This ensures that even if the application manually reclaims off-heap direct buffers immediately after sendMessage returns, the buffered message remains perfectly safe and valid for later transmission, resolving the outbound Use-After-Free/caching risk.

Author: kannanjgithub

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

@@ -991,7 +991,12 @@ public void sendMessage(InputStream message) {
 
         ExtProcStreamState state = extProcStreamState.get();
         if (state.isDraining() || state.isCompleted()) {
-          pendingDrainingMessages.add(message);
+          try {
+            ByteString copiedBody = ByteString.readFrom(message);
+            pendingDrainingMessages.add(new KnownLengthInputStream(copiedBody));
+          } catch (IOException e) {
+            rawCall.cancel("Failed to copy outbound message for buffering", e);
+          }
           return;
         }
       }
@@ -1301,7 +1306,12 @@ public void onMessage(InputStream message) {
 
         if (savedHeaders != null
             || dataPlaneClientCall.getExtProcStreamState().get().isDraining()) {
-          savedMessages.add(message);
+          try {
+            ByteString copiedBody = ByteString.readFrom(message);
+            savedMessages.add(new KnownLengthInputStream(copiedBody));
+          } catch (IOException e) {
+            rawCall.cancel("Failed to copy inbound message for buffering", e);
+          }
           return;
         }
       }