Implemented the missing EOS handling in handleResponseBodyResponse by half-closing the data plane rpc.

Author: kannanjgithub

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

@@ -849,16 +849,12 @@ private void handleResponseBodyResponse(io.envoyproxy.envoy.service.ext_proc.v3.
             if (!streamed.getBody().isEmpty()) {
               listener.onExternalBody(streamed.getBody());
             }
-            /*
             if (streamed.getEndOfStream() || streamed.getEndOfStreamWithoutMessage()) {
-              // Body stream from ext-proc finished, but we wait for rawCall.onClose to deliver final status.
-              // The filter would have already sent halfClose on the dataplane rpc in response to a 
-              // ProcessingResponse for a request, with end of stream indicated in that response. 
-              // So it now has to await for onClose() rather than do anything when 
-              // (streamed.getEndOfStream() || streamed.getEndOfStreamWithoutMessage()) 
-              // occurs in handleResponseBodyResponse.
+              if (requestSideClosed.compareAndSet(false, true)) {
+                super.halfClose();
+              }
+              listener.proceedWithClose();
             }
-            */
           }
         }
       }

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

@@ -1718,10 +1718,12 @@ public void onNext(ProcessingRequest request) {
         InProcessChannelBuilder.forName(uniqueDataPlaneServerName).directExecutor().build());
 
     final CountDownLatch appCloseLatch = new CountDownLatch(1);
+    final AtomicReference<Status> capturedStatus = new AtomicReference<>();
     CallOptions callOptions = CallOptions.DEFAULT.withExecutor(MoreExecutors.directExecutor());
     ClientCall<String, String> proxyCall = interceptor.interceptCall(METHOD_SAY_HELLO, callOptions, dataPlaneChannel);
     proxyCall.start(new ClientCall.Listener<String>() {
       @Override public void onClose(Status status, Metadata trailers) {
+        capturedStatus.set(status);
         appCloseLatch.countDown();
       }
     }, new Metadata());
@@ -1740,8 +1742,8 @@ public void onNext(ProcessingRequest request) {
 
     // Verify app listener notified
     assertThat(appCloseLatch.await(5, TimeUnit.SECONDS)).isTrue();
+    assertThat(capturedStatus.get().isOk()).isTrue();
 
-    proxyCall.cancel("Cleanup", null);
     channelManager.close();
   }