Address Review comments.

Author: kannanjgithub

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

@@ -17,7 +17,6 @@
 package io.grpc.testing.integration;
 
 import static com.google.common.truth.Truth.assertThat;
-import static io.grpc.testing.integration.TestCases.MCS_CS;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotEquals;
@@ -593,16 +592,14 @@ private void runTest(TestCases testCase) throws Exception {
         if (serverHostOverride != null) {
           channelBuilder.overrideAuthority(serverHostOverride);
         }
-        if (testCase.equals(MCS_CS.toString())) {
-          channelBuilder.disableServiceConfigLookUp();
-          try {
-            @SuppressWarnings("unchecked")
-            Map<String, ?> serviceConfigMap = (Map<String, ?>) JsonParser.parse(
-                "{\"connection_scaling\":{\"max_connections_per_subchannel\": 2}}");
-            channelBuilder.defaultServiceConfig(serviceConfigMap);
-          } catch (IOException e) {
-            throw new RuntimeException(e);
-          }
+        channelBuilder.disableServiceConfigLookUp();
+        try {
+          @SuppressWarnings("unchecked")
+          Map<String, ?> serviceConfigMap = (Map<String, ?>) JsonParser.parse(
+              "{\"connection_scaling\":{\"max_connections_per_subchannel\": 2}}");
+          channelBuilder.defaultServiceConfig(serviceConfigMap);
+        } catch (IOException e) {
+          throw new RuntimeException(e);
         }
         tester.testMcs(TestServiceGrpc.newStub(channelBuilder.build()));
         break;
@@ -1045,7 +1042,8 @@ public void testOrcaOob() throws Exception {
       streamObserver.onNext(StreamingOutputCallRequest.newBuilder()
           .setOrcaOobReport(answer2)
           .addResponseParameters(ResponseParameters.newBuilder().setSize(1).build()).build());
-      assertThat(streamingOutputCallResponseObserver.isCompleted).isTrue();
+      assertThat(streamingOutputCallResponseObserver.take())
+          .isInstanceOf(StreamingOutputCallResponse.class);
 
       for (i = 0; i < retryLimit; i++) {
         Thread.sleep(1000);
@@ -1055,6 +1053,8 @@ public void testOrcaOob() throws Exception {
         }
       }
       assertThat(i).isLessThan(retryLimit);
+      streamObserver.onCompleted();
+      assertThat(streamingOutputCallResponseObserver.verifiedCompleted()).isTrue();
     }
 
     @Override
@@ -1084,8 +1084,8 @@ protected int operationTimeoutMillis() {
 
     class StreamingOutputCallResponseObserver implements
         StreamObserver<StreamingOutputCallResponse> {
+      private final Object lastItem = new Object();
       private final BlockingQueue<Object> queue = new LinkedBlockingQueue<>();
-      private volatile boolean isCompleted = true;
 
       @Override
       public void onNext(StreamingOutputCallResponse value) {
@@ -1099,12 +1099,16 @@ public void onError(Throwable t) {
 
       @Override
       public void onCompleted() {
-        isCompleted = true;
+        queue.add(lastItem);
       }
 
       Object take() throws InterruptedException {
         return queue.take();
       }
+
+      boolean verifiedCompleted() throws InterruptedException {
+        return queue.take() == lastItem;
+      }
     }
 
     public void testMcs(TestServiceGrpc.TestServiceStub asyncStub) throws Exception {
@@ -1113,13 +1117,15 @@ public void testMcs(TestServiceGrpc.TestServiceStub asyncStub) throws Exception
       StreamObserver<StreamingOutputCallRequest> streamObserver1 =
           asyncStub.fullDuplexCall(responseObserver1);
       StreamingOutputCallRequest request = StreamingOutputCallRequest.newBuilder()
-          .setPayload(Payload.newBuilder().setBody(
-              ByteString.copyFromUtf8(MCS_CS.description())).build()).build();
+          .addResponseParameters(ResponseParameters.newBuilder()
+              .setSendClientSocketAddressInResponse(
+                  Messages.BoolValue.newBuilder().setValue(true).build())
+              .build())
+          .build();
       streamObserver1.onNext(request);
       Object responseObj = responseObserver1.take();
       StreamingOutputCallResponse callResponse = (StreamingOutputCallResponse) responseObj;
-      String clientSocketAddressInCall1 = new String(callResponse.getPayload().getBody()
-          .toByteArray(), UTF_8);
+      String clientSocketAddressInCall1 = callResponse.getClientSocketAddress();
       assertThat(clientSocketAddressInCall1).isNotEmpty();
 
       StreamingOutputCallResponseObserver responseObserver2 =
@@ -1128,30 +1134,30 @@ public void testMcs(TestServiceGrpc.TestServiceStub asyncStub) throws Exception
           asyncStub.fullDuplexCall(responseObserver2);
       streamObserver2.onNext(request);
       callResponse = (StreamingOutputCallResponse) responseObserver2.take();
-      String clientSocketAddressInCall2 =
-          new String(callResponse.getPayload().getBody().toByteArray(), UTF_8);
+      String clientSocketAddressInCall2 = callResponse.getClientSocketAddress();
 
       assertThat(clientSocketAddressInCall1).isEqualTo(clientSocketAddressInCall2);
 
       // 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 =
+      StreamingOutputCallResponseObserver responseObserver3 =
           new StreamingOutputCallResponseObserver();
       StreamObserver<StreamingOutputCallRequest> streamObserver3 =
           asyncStub.fullDuplexCall(responseObserver3);
       streamObserver3.onNext(request);
       callResponse = (StreamingOutputCallResponse) responseObserver3.take();
-      String clientSocketAddressInCall3 =
-          new String(callResponse.getPayload().getBody().toByteArray(), UTF_8);
+      String clientSocketAddressInCall3 = callResponse.getClientSocketAddress();
 
-      assertThat(clientSocketAddressInCall3).isNotEqualTo(clientSocketAddressInCall1);*/
+      // This assertion is currently failing because connection scaling when MCS limit has been
+      // reached is not yet implemented in gRPC Java.
+      assertThat(clientSocketAddressInCall3).isNotEqualTo(clientSocketAddressInCall1);
 
       streamObserver1.onCompleted();
-      assertThat(responseObserver1.isCompleted).isTrue();
+      assertThat(responseObserver1.verifiedCompleted()).isTrue();
       streamObserver2.onCompleted();
-      assertThat(responseObserver2.isCompleted).isTrue();
-      /*streamObserver3.onCompleted();
-      assertThat(responseObserver3.isCompleted).isTrue();*/
+      assertThat(responseObserver2.verifiedCompleted()).isTrue();
+      streamObserver3.onCompleted();
+      assertThat(responseObserver3.verifiedCompleted()).isTrue();
     }
   }
 

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

@@ -17,8 +17,6 @@
 package io.grpc.testing.integration;
 
 import static io.grpc.Grpc.TRANSPORT_ATTR_REMOTE_ADDR;
-import static io.grpc.testing.integration.TestCases.MCS_CS;
-import static java.nio.charset.StandardCharsets.UTF_8;
 
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Queues;
@@ -54,6 +52,7 @@
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Queue;
@@ -243,20 +242,27 @@ public void onNext(StreamingOutputCallRequest request) {
               .asRuntimeException());
           return;
         }
-        if (new String(request.getPayload().getBody().toByteArray(), UTF_8)
-            .equals(MCS_CS.description())) {
-          SocketAddress peerAddress = PEER_ADDRESS_CONTEXT_KEY.get();
-          ByteString payload = ByteString.copyFromUtf8(peerAddress.toString());
-          StreamingOutputCallResponse.Builder responseBuilder =
-              StreamingOutputCallResponse.newBuilder();
-          responseBuilder.setPayload(
-              Payload.newBuilder()
-                  .setBody(payload));
-          responseObserver.onNext(responseBuilder.build());
+        if (whetherSendClientSocketAddressInResponse(request)) {
+          responseObserver.onNext(
+              StreamingOutputCallResponse.newBuilder()
+                  .setClientSocketAddress(PEER_ADDRESS_CONTEXT_KEY.get().toString())
+                  .build());
+          return;
         }
         dispatcher.enqueue(toChunkQueue(request));
       }
 
+      private boolean whetherSendClientSocketAddressInResponse(StreamingOutputCallRequest request) {
+        Iterator<ResponseParameters> responseParametersIterator =
+            request.getResponseParametersList().iterator();
+        while (responseParametersIterator.hasNext()) {
+          if (responseParametersIterator.next().getSendClientSocketAddressInResponse().getValue()) {
+            return true;
+          }
+        }
+        return false;
+      }
+
       @Override
       public void onCompleted() {
         if (oobTestLocked) {

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

@@ -75,7 +75,7 @@ public void run() {
   private int port = 8080;
   private boolean useTls = true;
   private boolean useAlts = false;
-  private boolean useMcs = false;
+  private boolean setMcsLimit = false;
 
   private ScheduledExecutorService executor;
   private Server server;
@@ -119,8 +119,9 @@ void parseArgs(String[] args) {
           usage = true;
           break;
         }
-      } else if ("use_mcs".equals(key)) {
-        useMcs = Boolean.parseBoolean(value);
+      } else if ("set_max_concurrent_streams_limit".equals(key)) {
+        setMcsLimit = Boolean.parseBoolean(value);
+        // TODO: Make Netty server builder usable for IPV6 as well (not limited to MCS handling)
         addressType = Util.AddressType.IPV4; // To use NettyServerBuilder
       } else {
         System.err.println("Unknown argument: " + key);
@@ -145,6 +146,8 @@ void parseArgs(String[] args) {
               + "\n                        for testing. Only effective when --use_alts=true."
               + "\n  --address_type=IPV4|IPV6|IPV4_IPV6"
               + "\n                        What type of addresses to listen on. Default IPV4_IPV6"
+              + "\n  --set_max_concurrent_streams_limit"
+              + "\n                        Whether to set the maximum concurrent streams limit"
       );
       System.exit(1);
     }
@@ -190,7 +193,7 @@ void start() throws Exception {
         if (v4Address != null && !v4Address.equals(localV4Address)) {
           ((NettyServerBuilder) serverBuilder).addListenAddress(v4Address);
         }
-        if (useMcs) {
+        if (setMcsLimit) {
           ((NettyServerBuilder) serverBuilder).maxConcurrentCallsPerConnection(2);
         }
         break;

interop-testing/src/main/proto/grpc/testing/messages.proto

@@ -159,6 +159,10 @@ message ResponseParameters {
   // implement the full compression tests by introspecting the call to verify
   // the response's compression status.
   BoolValue compressed = 3;
+
+  // Whether to request the server to send the requesting client's socket
+  // address in the response.
+  BoolValue send_client_socket_address_in_response = 4;
 }
 
 // Server-streaming request.
@@ -186,6 +190,9 @@ message StreamingOutputCallRequest {
 message StreamingOutputCallResponse {
   // Payload to increase response size.
   Payload payload = 1;
+
+  // The client's socket address if requested.
+  string client_socket_address = 2;
 }
 
 // For reconnect interop test only.