xds: fix xdsClient resource not exist for invalid resource, fix xdsServerWrapper start on resource not exist

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

@@ -2214,11 +2214,13 @@ private void handleResourceUpdate(
             }
             retainedResources.add(edsName);
           }
-          continue;
+        } else if (invalidResources.contains(resourceName)) {
+          subscriber.onError(Status.UNAVAILABLE.withDescription(errorDetail));
+        } else {
+          // For State of the World services, notify watchers when their watched resource is missing
+          // from the ADS update.
+          subscriber.onAbsent();
         }
-        // For State of the World services, notify watchers when their watched resource is missing
-        // from the ADS update.
-        subscriber.onAbsent();
       }
     }
     // LDS/CDS responses represents the state of the world, RDS/EDS resources not referenced in

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

@@ -571,10 +571,6 @@ private void handleConfigNotFound(StatusException exception) {
       for (SslContextProviderSupplier s: toRelease) {
         s.close();
       }
-      if (!initialStarted) {
-        initialStarted = true;
-        initialStartFuture.set(exception);
-      }
       if (restartTimer != null) {
         restartTimer.cancel();
       }

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

@@ -1563,12 +1563,16 @@ public void cdsResponseErrorHandling_badUpstreamTlsContext() {
     call.sendResponse(CDS, clusters, VERSION_1, "0000");
 
     // The response NACKed with errors indicating indices of the failed resources.
-    call.verifyRequestNack(CDS, CDS_RESOURCE, "", "0000", NODE, ImmutableList.of(
-        "CDS response Cluster 'cluster.googleapis.com' validation error: "
+    String errorMsg =  "CDS response Cluster 'cluster.googleapis.com' validation error: "
             + "Cluster cluster.googleapis.com: malformed UpstreamTlsContext: "
             + "io.grpc.xds.ClientXdsClient$ResourceInvalidException: "
-            + "ca_certificate_provider_instance is required in upstream-tls-context"));
-    verifyNoInteractions(cdsResourceWatcher);
+            + "ca_certificate_provider_instance is required in upstream-tls-context";
+    call.verifyRequestNack(CDS, CDS_RESOURCE, "", "0000", NODE, ImmutableList.of(errorMsg));
+    ArgumentCaptor<Status> captor = ArgumentCaptor.forClass(Status.class);
+    verify(cdsResourceWatcher).onError(captor.capture());
+    Status errorStatus = captor.getValue();
+    assertThat(errorStatus.getCode()).isEqualTo(Status.UNAVAILABLE.getCode());
+    assertThat(errorStatus.getDescription()).isEqualTo(errorMsg);
   }
 
   /**
@@ -1587,10 +1591,14 @@ public void cdsResponseErrorHandling_badTransportSocketName() {
     call.sendResponse(CDS, clusters, VERSION_1, "0000");
 
     // The response NACKed with errors indicating indices of the failed resources.
-    call.verifyRequestNack(CDS, CDS_RESOURCE, "", "0000", NODE, ImmutableList.of(
-        "CDS response Cluster 'cluster.googleapis.com' validation error: "
-            + "transport-socket with name envoy.transport_sockets.bad not supported."));
-    verifyNoInteractions(cdsResourceWatcher);
+    String errorMsg = "CDS response Cluster 'cluster.googleapis.com' validation error: "
+        + "transport-socket with name envoy.transport_sockets.bad not supported.";
+    call.verifyRequestNack(CDS, CDS_RESOURCE, "", "0000", NODE, ImmutableList.of(errorMsg));
+    ArgumentCaptor<Status> captor = ArgumentCaptor.forClass(Status.class);
+    verify(cdsResourceWatcher).onError(captor.capture());
+    Status errorStatus = captor.getValue();
+    assertThat(errorStatus.getCode()).isEqualTo(Status.UNAVAILABLE.getCode());
+    assertThat(errorStatus.getDescription()).isEqualTo(errorMsg);
   }
 
   @Test
@@ -2438,10 +2446,15 @@ public void serverSideListenerResponseErrorHandling_badDownstreamTlsContext() {
     List<Any> listeners = ImmutableList.of(Any.pack(listener));
     call.sendResponse(ResourceType.LDS, listeners, "0", "0000");
     // The response NACKed with errors indicating indices of the failed resources.
-    call.verifyRequestNack(LDS, LISTENER_RESOURCE, "", "0000", NODE, ImmutableList.of(
-            "LDS response Listener \'grpc/server?xds.resource.listening_address=0.0.0.0:7000\' "
-                + "validation error: common-tls-context is required in downstream-tls-context"));
-    verifyNoInteractions(ldsResourceWatcher);
+    String errorMsg = "LDS response Listener \'grpc/server?xds.resource.listening_address="
+        + "0.0.0.0:7000\' validation error: "
+        + "common-tls-context is required in downstream-tls-context";
+    call.verifyRequestNack(LDS, LISTENER_RESOURCE, "", "0000", NODE, ImmutableList.of(errorMsg));
+    ArgumentCaptor<Status> captor = ArgumentCaptor.forClass(Status.class);
+    verify(ldsResourceWatcher).onError(captor.capture());
+    Status errorStatus = captor.getValue();
+    assertThat(errorStatus.getCode()).isEqualTo(Status.UNAVAILABLE.getCode());
+    assertThat(errorStatus.getDescription()).isEqualTo(errorMsg);
   }
 
   @Test
@@ -2462,11 +2475,16 @@ public void serverSideListenerResponseErrorHandling_badTransportSocketName() {
     List<Any> listeners = ImmutableList.of(Any.pack(listener));
     call.sendResponse(ResourceType.LDS, listeners, "0", "0000");
     // The response NACKed with errors indicating indices of the failed resources.
+    String errorMsg = "LDS response Listener \'grpc/server?xds.resource.listening_address="
+        + "0.0.0.0:7000\' validation error: "
+        + "transport-socket with name envoy.transport_sockets.bad1 not supported.";
     call.verifyRequestNack(LDS, LISTENER_RESOURCE, "", "0000", NODE, ImmutableList.of(
-        "LDS response Listener \'grpc/server?xds.resource.listening_address=0.0.0.0:7000\' "
-            + "validation error: "
-            + "transport-socket with name envoy.transport_sockets.bad1 not supported."));
-    verifyNoInteractions(ldsResourceWatcher);
+        errorMsg));
+    ArgumentCaptor<Status> captor = ArgumentCaptor.forClass(Status.class);
+    verify(ldsResourceWatcher).onError(captor.capture());
+    Status errorStatus = captor.getValue();
+    assertThat(errorStatus.getCode()).isEqualTo(Status.UNAVAILABLE.getCode());
+    assertThat(errorStatus.getDescription()).isEqualTo(errorMsg);
   }
 
   private DiscoveryRpcCall startResourceWatcher(

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

@@ -72,6 +72,8 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -192,6 +194,9 @@ public void run() {
       fail("Start should throw exception");
     } catch (ExecutionException ex) {
       assertThat(ex.getCause()).isInstanceOf(IOException.class);
+      fail("Start should not fail");
+    } catch (TimeoutException ex) {
+      assertThat(start.isDone()).isFalse();
     }
     assertThat(selectorManager.getSelectorToUpdateSelector()).isSameInstanceAs(NO_FILTER_CHAIN);
   }
@@ -216,6 +221,9 @@ public void run() {
       fail("Start should throw exception");
     } catch (ExecutionException ex) {
       assertThat(ex.getCause()).isInstanceOf(IOException.class);
+      fail("Start should not fail");
+    } catch (TimeoutException ex) {
+      assertThat(start.isDone()).isFalse();
     }
     assertThat(selectorManager.getSelectorToUpdateSelector()).isSameInstanceAs(NO_FILTER_CHAIN);
   }
@@ -240,6 +248,9 @@ public void run() {
       fail("Start should throw exception");
     } catch (ExecutionException ex) {
       assertThat(ex.getCause()).isInstanceOf(IOException.class);
+      fail("Start should not fail");
+    } catch (TimeoutException ex) {
+      assertThat(start.isDone()).isFalse();
     }
     assertThat(selectorManager.getSelectorToUpdateSelector()).isSameInstanceAs(NO_FILTER_CHAIN);
   }

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

@@ -74,6 +74,7 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 import org.junit.Before;
 import org.junit.Rule;
@@ -261,9 +262,13 @@ public void run() {
     xdsClient.ldsWatcher.onResourceDoesNotExist(ldsResource);
     try {
       start.get(5000, TimeUnit.MILLISECONDS);
-      fail("Start should throw exception");
+      fail("server should not start() successfully.");
     } catch (ExecutionException ex) {
       assertThat(ex.getCause()).isInstanceOf(IOException.class);
+      fail("server start() should not fail.");
+    } catch (TimeoutException ex) {
+      // expect to block here.
+      assertThat(start.isDone()).isFalse();
     }
     verify(mockBuilder, times(1)).build();
     verify(mockServer, never()).start();
@@ -602,9 +607,10 @@ public void run() {
     xdsClient.ldsWatcher.onResourceDoesNotExist(ldsResource);
     try {
       start.get(5000, TimeUnit.MILLISECONDS);
-      fail("Start should throw exception");
-    } catch (ExecutionException ex) {
-      assertThat(ex.getCause()).isInstanceOf(IOException.class);
+      fail("server should not start()");
+    } catch (TimeoutException ex) {
+      // expect to block here.
+      assertThat(start.isDone()).isFalse();
     }
     verify(listener, times(1)).onNotServing(any(StatusException.class));
     verify(mockBuilder, times(1)).build();
@@ -627,6 +633,13 @@ public void run() {
     assertThat(sslSupplier0.isShutdown()).isTrue();
     xdsClient.deliverRdsUpdate("rds",
             Collections.singletonList(createVirtualHost("virtual-host-1")));
+    try {
+      start.get(5000, TimeUnit.MILLISECONDS);
+      fail("Start should throw exception");
+    } catch (ExecutionException ex) {
+      assertThat(ex.getCause()).isInstanceOf(IOException.class);
+      assertThat(ex.getCause().getMessage()).isEqualTo("error!");
+    }
     RdsResourceWatcher saveRdsWatcher = xdsClient.rdsWatchers.get("rds");
     assertThat(executor.forwardNanos(RETRY_DELAY_NANOS)).isEqualTo(1);
     verify(mockBuilder, times(1)).build();