The test failed with "too many messages" error due to sending an empty body message to the data plane server during the "half-close" phase . For a unary RPC, this was interpreted as a second request message, which is invalid. I've updated handleRequestBodyResponse and onExternalBody to only call super.sendMessage() or super.onMessage() if the body content is non-empty. This prevents the redundant empty message from being sent to the data plane while still allowing the external processor to signal the end of the stream.

Author: kannanjgithub

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

@@ -549,9 +549,10 @@ private void handleRequestBodyResponse(io.envoyproxy.envoy.service.ext_proc.v3.B
           io.envoyproxy.envoy.service.ext_proc.v3.BodyMutation mutation = bodyResponse.getResponse().getBodyMutation();
           if (mutation.hasBody()) {
             byte[] mutatedBody = mutation.getBody().toByteArray();
-            super.sendMessage(new ByteArrayInputStream(mutatedBody));
+            if (mutatedBody.length > 0) {
+              super.sendMessage(new ByteArrayInputStream(mutatedBody));
+            }
           } else if (mutation.getClearBody()) {
-            // "clear_body" means we should send an empty message.
             super.sendMessage(new ByteArrayInputStream(new byte[0]));
           }
           // If body mutation is present but has no body and clear_body is false, do nothing.
@@ -718,7 +719,9 @@ void proceedWithClose() {
       }
 
       void onExternalBody(com.google.protobuf.ByteString body) {
-         super.onMessage(body.newInput());
+         if (body.size() > 0) {
+           super.onMessage(body.newInput());
+         }
       }
 
       void unblockAfterStreamComplete() {

xds/src/test/java/io/grpc/xds/ExternalProcessorFilterTest.java

@@ -168,7 +168,7 @@ public void requestHeadersMutated() throws Exception {
     ExternalProcessorFilterConfig filterConfig = createFilterConfig();
 
     // Manually create the interceptor using the test-friendly constructor
-    CachedChannelManager testChannelManager = new CachedChannelManager(config -> 
+    CachedChannelManager testChannelManager = new CachedChannelManager(config ->
         grpcCleanup.register(InProcessChannelBuilder.forName(extProcServerName).directExecutor().build())
     );
     ClientInterceptor interceptor = new ExternalProcessorInterceptor(filterConfig, testChannelManager);