Extend response drain test to verify handshake completion.

Extend givenResponseDrainActive_whenAppRequestsMessages_thenRequestsBuffered to complete the response draining handshake and verify that buffered requests are then successfully sent to the data plane.

TAG=agy
CONV=2c1e4760-c239-4698-810a-162bf10fccc4

Author: kannanjgithub

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

@@ -9022,13 +9022,17 @@ public void givenResponseDrainActive_whenAppRequestsMessages_thenRequestsBuffere
     final CountDownLatch drainSentLatch = new CountDownLatch(1);
     final CountDownLatch headersReceivedLatch = new CountDownLatch(1);
     final CountDownLatch sendDrainLatch = new CountDownLatch(1);
+    final CountDownLatch filterSentDrainCompleteLatch = new CountDownLatch(1);
+    final AtomicReference<StreamObserver<ProcessingResponse>> responseObserverRef =
+        new AtomicReference<>();
     // External Processor Server
     ExternalProcessorGrpc.ExternalProcessorImplBase extProcImpl;
     extProcImpl = new ExternalProcessorGrpc.ExternalProcessorImplBase() {
       @Override
       @SuppressWarnings("unchecked")
       public StreamObserver<ProcessingRequest> process(
           final StreamObserver<ProcessingResponse> responseObserver) {
+        responseObserverRef.set(responseObserver);
         ((ServerCallStreamObserver<ProcessingResponse>) responseObserver).request(100);
         return new StreamObserver<ProcessingRequest>() {
           @Override
@@ -9048,6 +9052,10 @@ public void onNext(ProcessingRequest request) {
                   System.out.println("JetskiDebug: error in mock server: " + e);
                 }
               }).start();
+            } else if (request.hasResponseBody()) {
+              if (request.getResponseBody().getDrainComplete()) {
+                filterSentDrainCompleteLatch.countDown();
+              }
             }
           }
 
@@ -9123,6 +9131,27 @@ public void request(int numMessages) {
 
     // Verify requests are buffered and not sent to data plane
     assertThat(dataPlaneRequestCount.get()).isEqualTo(0);
+
+    // Wait for filter to send drain_complete to mock server
+    assertThat(filterSentDrainCompleteLatch.await(5, TimeUnit.SECONDS)).isTrue();
+
+    // Echo drain_complete back to filter to complete handshake
+    synchronized (responseObserverRef.get()) {
+      responseObserverRef.get().onNext(ProcessingResponse.newBuilder()
+          .setResponseBody(BodyResponse.newBuilder()
+              .setResponse(CommonResponse.newBuilder()
+                  .setBodyMutation(BodyMutation.newBuilder()
+                      .setStreamedResponse(StreamedBodyResponse.newBuilder()
+                          .setDrainComplete(true)
+                          .build())
+                      .build())
+                  .build())
+              .build())
+          .build());
+    }
+
+    // Verify requests are now drained to data plane
+    assertThat(dataPlaneRequestCount.get()).isEqualTo(3);
 
     proxyCall.cancel("Cleanup", null);
     channelManager.close();