bug fix to requested resource does not exist handling

cache/src/main/java/io/envoyproxy/controlplane/cache/SimpleCache.java

@@ -377,10 +377,9 @@ protected void respondWithSpecificOrder(T group,
                   version);
             }
 
-            respond(watch, snapshot, group);
-
-            // Discard the watch. A new watch will be created for future snapshots once envoy ACKs the response.
-            return true;
+            // If we respond with an actual message, we can proceed to discard the watch. 
+            // A new watch will be created for future snapshots once envoy ACKs the response.
+            return respond(watch, snapshot, group);
           }
 
           // Do not discard the watch. The request version is the same as the snapshot version, so we wait to respond.

cache/src/test/java/io/envoyproxy/controlplane/cache/v3/SimpleCacheTest.java

@@ -476,6 +476,64 @@ public void watchIsLeftOpenIfNotRespondedImmediately() {
     assertThatWatchIsOpenWithNoResponses(new WatchAndTracker(watch, responseTracker));
   }
 
+  @Test
+  public void watchIsLeftOpenIfNotRespondedImmediatelyAndThroughSubsequentSetEmptySnapshots() {
+    SimpleCache<String> cache = new SimpleCache<>(new SingleNodeGroup());
+    cache.setSnapshot(SingleNodeGroup.GROUP, Snapshot.create(
+        ImmutableList.of(), ImmutableList.of(), ImmutableList.of(), ImmutableList.of(), ImmutableList.of(), VERSION1));
+
+    ResponseTracker responseTracker = new ResponseTracker();
+    Watch watch = cache.createWatch(
+        true,
+        XdsRequest.create(DiscoveryRequest.newBuilder()
+            .setNode(Node.getDefaultInstance())
+            .setTypeUrl(ROUTE_TYPE_URL)
+            .addAllResourceNames(Collections.singleton(ROUTE_NAME))
+            .build()),
+        Collections.singleton(ROUTE_NAME),
+        responseTracker);
+
+    assertThatWatchIsOpenWithNoResponses(new WatchAndTracker(watch, responseTracker));
+    assertThat(cache.statusInfo(SingleNodeGroup.GROUP).numWatches()).isEqualTo(1);
+
+    cache.setSnapshot(SingleNodeGroup.GROUP, Snapshot.create(
+        ImmutableList.of(), ImmutableList.of(), ImmutableList.of(), ImmutableList.of(), ImmutableList.of(), VERSION2));
+
+    assertThatWatchIsOpenWithNoResponses(new WatchAndTracker(watch, responseTracker));
+    assertThat(cache.statusInfo(SingleNodeGroup.GROUP).numWatches()).isEqualTo(1);
+  }
+
+  @Test
+  public void watchIsLeftOpenIfNotRespondedImmediatelyAndLaterSetSnapshotSendsUpdate() {
+    SimpleCache<String> cache = new SimpleCache<>(new SingleNodeGroup());
+    cache.setSnapshot(SingleNodeGroup.GROUP, Snapshot.create(
+        ImmutableList.of(), ImmutableList.of(), ImmutableList.of(), ImmutableList.of(), ImmutableList.of(), VERSION1));
+
+    ResponseTracker responseTracker = new ResponseTracker();
+    Watch watch = cache.createWatch(
+        true,
+        XdsRequest.create(DiscoveryRequest.newBuilder()
+            .setNode(Node.getDefaultInstance())
+            .setTypeUrl(ROUTE_TYPE_URL)
+            .addAllResourceNames(SNAPSHOT1.resources(ROUTE_TYPE_URL).keySet())
+            .build()),
+        Collections.emptySet(),
+        responseTracker);
+
+    assertThatWatchIsOpenWithNoResponses(new WatchAndTracker(watch, responseTracker));
+    assertThat(cache.statusInfo(SingleNodeGroup.GROUP).numWatches()).isEqualTo(1);
+
+    cache.setSnapshot(SingleNodeGroup.GROUP, Snapshot.create(
+        ImmutableList.of(), ImmutableList.of(), ImmutableList.of(), ImmutableList.of(), ImmutableList.of(), VERSION2));
+
+    assertThatWatchIsOpenWithNoResponses(new WatchAndTracker(watch, responseTracker));
+    assertThat(cache.statusInfo(SingleNodeGroup.GROUP).numWatches()).isEqualTo(1);
+
+    cache.setSnapshot(SingleNodeGroup.GROUP, SNAPSHOT1);
+
+    assertThatWatchReceivesSnapshot(new WatchAndTracker(watch, responseTracker), SNAPSHOT1);
+  }
+
   @Test
   public void getSnapshot() {
     SimpleCache<String> cache = new SimpleCache<>(new SingleNodeGroup());