Merge branch 'client-watcher-changes-2' of https://github.com/shivaspeaks/grpc-java

Author: shivaspeaks

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

@@ -4580,9 +4580,15 @@ public void serverFailureMetricReport_forRetryAndBackoff() {
         argThat(status -> status.getCode() == Code.DEADLINE_EXCEEDED));
     rdsWatcherInOrder.verify(rdsResourceWatcher).onAmbientError(
         argThat(status -> status.getCode() == Code.DEADLINE_EXCEEDED));
-    // Server Failure metric will not be reported, as stream is closed with an error after receiving
+    cdsWatcherInOrder.verify(cdsResourceWatcher).onResourceChanged(
+        argThat(statusOr -> !statusOr.hasValue()
+            && statusOr.getStatus().getCode() == Code.DEADLINE_EXCEEDED));
+    edsWatcherInOrder.verify(edsResourceWatcher).onResourceChanged(
+        argThat(statusOr -> !statusOr.hasValue()
+            && statusOr.getStatus().getCode() == Code.DEADLINE_EXCEEDED));
+    // Server Failure metric is now reported, as stream is closed with an error after receiving
     // a response
-    verifyServerFailureCount(2, 1, xdsServerInfo.target());
+    verifyServerFailureCount(3, 1, xdsServerInfo.target());
 
     // Reset backoff sequence and retry after backoff.
     inOrder.verify(backoffPolicyProvider).get();
@@ -4597,7 +4603,7 @@ public void serverFailureMetricReport_forRetryAndBackoff() {
     call.sendError(Status.UNAVAILABLE.asException());
     ldsWatcherInOrder.verify(ldsResourceWatcher).onAmbientError(
         argThat(status -> status.getCode() == Code.UNAVAILABLE));
-    verifyServerFailureCount(3, 1, xdsServerInfo.target());
+    verifyServerFailureCount(4, 1, xdsServerInfo.target());
 
     // Retry after backoff.
     inOrder.verify(backoffPolicy2).nextBackoffNanos();
@@ -4610,12 +4616,10 @@ public void serverFailureMetricReport_forRetryAndBackoff() {
     List<Any> clusters = ImmutableList.of(FAILING_ANY, testClusterRoundRobin);
     call.sendResponse(CDS, clusters, VERSION_1, "0000");
     call.sendCompleted();
-    // Server Failure metric will not be reported once again, as stream is closed after receiving a
-    // response
-    verifyServerFailureCount(3, 1, xdsServerInfo.target());
+    // Server Failure metric will not be reported, as stream is closed gracefully.
+    verifyServerFailureCount(4, 1, xdsServerInfo.target());
   }
 
-
   private XdsClientImpl createXdsClient(String serverUri) {
     BootstrapInfo bootstrapInfo = buildBootStrap(serverUri);
     return new XdsClientImpl(

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

@@ -461,18 +461,14 @@ public XdsTransport create(Bootstrapper.ServerInfo serverInfo) {
   public void connect_then_mainServerRestart_fallbackServerdown() {
     mainXdsServer.restartXdsServer();
     xdsClient = xdsClientPool.getObject();
-
     xdsClient.watchXdsResource(XdsListenerResource.getInstance(), MAIN_SERVER, ldsWatcher);
 
     verify(ldsWatcher, timeout(5000)).onResourceChanged(
         argThat(statusOr -> statusOr.hasValue() && statusOr.getValue().equals(
             LdsUpdate.forApiListener(MAIN_HTTP_CONNECTION_MANAGER))));
-
     mainXdsServer.getServer().shutdownNow();
     fallbackServer.getServer().shutdownNow();
-
     xdsClient.watchXdsResource(XdsClusterResource.getInstance(), CLUSTER_NAME, cdsWatcher);
-
     mainXdsServer.restartXdsServer();
 
     verify(cdsWatcher, timeout(5000)).onResourceChanged(