Address review comments.

Author: kannanjgithub

android-interop-testing/build.gradle

@@ -17,7 +17,6 @@ android {
                 srcDirs += "${projectDir}/../interop-testing/src/main/java/"
                 setIncludes(["io/grpc/android/integrationtest/**",
                              "io/grpc/testing/integration/AbstractInteropTest.java",
-                             "io/grpc/testing/integration/TestCases.java",
                              "io/grpc/testing/integration/TestServiceImpl.java",
                              "io/grpc/testing/integration/Util.java"])
             }

interop-testing/src/main/java/io/grpc/testing/integration/TestServiceClient.java

@@ -1020,8 +1020,9 @@ public void testOrcaOob() throws Exception {
           .build();
 
       final int retryLimit = 5;
+      final Object lastItem = new Object();
       StreamingOutputCallResponseObserver streamingOutputCallResponseObserver =
-          new StreamingOutputCallResponseObserver();
+          new StreamingOutputCallResponseObserver(lastItem);
       StreamObserver<StreamingOutputCallRequest> streamObserver =
           asyncStub.fullDuplexCall(streamingOutputCallResponseObserver);
 
@@ -1054,7 +1055,7 @@ public void testOrcaOob() throws Exception {
       }
       assertThat(i).isLessThan(retryLimit);
       streamObserver.onCompleted();
-      assertThat(streamingOutputCallResponseObserver.verifiedCompleted()).isTrue();
+      assertThat(streamingOutputCallResponseObserver.take()).isSameInstanceAs(lastItem);
     }
 
     @Override
@@ -1084,9 +1085,13 @@ protected int operationTimeoutMillis() {
 
     class StreamingOutputCallResponseObserver implements
         StreamObserver<StreamingOutputCallResponse> {
-      private final Object lastItem = new Object();
+      private final Object lastItem;
       private final BlockingQueue<Object> queue = new LinkedBlockingQueue<>();
 
+      public StreamingOutputCallResponseObserver(Object lastItem) {
+        this.lastItem = lastItem;
+      }
+
       @Override
       public void onNext(StreamingOutputCallResponse value) {
         queue.add(value);
@@ -1105,15 +1110,12 @@ public void onCompleted() {
       Object take() throws InterruptedException {
         return queue.take();
       }
-
-      boolean verifiedCompleted() throws InterruptedException {
-        return queue.take() == lastItem;
-      }
     }
 
     public void testMcs(TestServiceGrpc.TestServiceStub asyncStub) throws Exception {
+      final Object lastItem = new Object();
       StreamingOutputCallResponseObserver responseObserver1 =
-          new StreamingOutputCallResponseObserver();
+          new StreamingOutputCallResponseObserver(lastItem);
       StreamObserver<StreamingOutputCallRequest> streamObserver1 =
           asyncStub.fullDuplexCall(responseObserver1);
       StreamingOutputCallRequest request = StreamingOutputCallRequest.newBuilder()
@@ -1129,7 +1131,7 @@ public void testMcs(TestServiceGrpc.TestServiceStub asyncStub) throws Exception
       assertThat(clientSocketAddressInCall1).isNotEmpty();
 
       StreamingOutputCallResponseObserver responseObserver2 =
-          new StreamingOutputCallResponseObserver();
+          new StreamingOutputCallResponseObserver(lastItem);
       StreamObserver<StreamingOutputCallRequest> streamObserver2 =
           asyncStub.fullDuplexCall(responseObserver2);
       streamObserver2.onNext(request);
@@ -1141,7 +1143,7 @@ public void testMcs(TestServiceGrpc.TestServiceStub asyncStub) throws Exception
       // The first connection is at max rpc call count of 2, so the 3rd rpc will cause a new
       // connection to be created in the same subchannel and not get queued.
       StreamingOutputCallResponseObserver responseObserver3 =
-          new StreamingOutputCallResponseObserver();
+          new StreamingOutputCallResponseObserver(lastItem);
       StreamObserver<StreamingOutputCallRequest> streamObserver3 =
           asyncStub.fullDuplexCall(responseObserver3);
       streamObserver3.onNext(request);
@@ -1153,11 +1155,11 @@ public void testMcs(TestServiceGrpc.TestServiceStub asyncStub) throws Exception
       assertThat(clientSocketAddressInCall3).isNotEqualTo(clientSocketAddressInCall1);
 
       streamObserver1.onCompleted();
-      assertThat(responseObserver1.verifiedCompleted()).isTrue();
       streamObserver2.onCompleted();
-      assertThat(responseObserver2.verifiedCompleted()).isTrue();
       streamObserver3.onCompleted();
-      assertThat(responseObserver3.verifiedCompleted()).isTrue();
+      assertThat(responseObserver1.take()).isSameInstanceAs(lastItem);
+      assertThat(responseObserver2.take()).isSameInstanceAs(lastItem);
+      assertThat(responseObserver3.take()).isSameInstanceAs(lastItem);
     }
   }