xds: skip DiscoveryRequest for unsubscribed types on stream ready (#12782)

## TL;DR

When the bootstrap declares more than one xDS server (e.g. a default
server for LDS/CDS plus an authority-specific EDS-only server),
grpc-java was sending CDS/LDS DiscoveryRequests to the EDS-only server
too. That server replies `UNIMPLEMENTED`, which tears down the stream
and EDS data never arrives. Fix: skip DiscoveryRequests for resource
types we don't actually subscribe to on a given server.

## Problem

Under [A47 — xDS
Federation](https://github.com/grpc/proposal/blob/master/A47-xds-federation.md),
authorities can declare their own `xds_servers` block in the bootstrap.
When an ADS stream is opened to an authority-specific server,
`ControlPlaneClient.adjustResourceSubscription` sends a
`DiscoveryRequest` for every **globally-subscribed** resource type —
even types that have no subscription for *this* server. The empty
request still carries a `type_url`, and an authority-specific server
(e.g. an EDS-only control plane) may reject it with `UNIMPLEMENTED`,
which tears down the entire stream before the legitimate request that
follows can complete.

## Reproduce

Bootstrap declares two servers — call them control-plane-A (handles
LDS/CDS for authority A) and control-plane-B (handles EDS only for
authority B).

A grpc-java channel that resolves through LDS → CDS in authority A and
EDS in authority B opens an ADS stream to control-plane-B. When that
stream becomes ready,
[`sendDiscoveryRequests`](https://github.com/grpc/grpc-java/blob/master/xds/src/main/java/io/grpc/xds/client/ControlPlaneClient.java)
iterates `resourceStore.getSubscribedResourceTypesWithTypeUrl()` — which
returns **all three** types (Listener, Cluster, Endpoint) — and calls
`adjustResourceSubscription` for each. For Listener and Cluster,
`getSubscribedResources(serverInfo, type)` returns null/empty, but the
request is still sent on the wire:

```
io.grpc.StatusRuntimeException: UNAVAILABLE: Error retrieving EDS resource ...:
  UNIMPLEMENTED. Details: Watches for type type.googleapis.com/envoy.config.cluster.v3.Cluster
  are not supported in this service
```

The Cluster (CDS) request reaches control-plane-B, gets rejected, and
the stream goes into backoff with no EDS data ever delivered. grpc-go on
the same bootstrap works fine against the same server, which pointed at
the asymmetry.

## Fix

In `adjustResourceSubscription`, return early when both:

1. `resources` is `null` (the store reports no subscription for this
type on this server), and
2. No DiscoveryRequest of this type has been sent on the current stream
(`!adsStream.sentTypes.contains(resourceType)`).

Per the `ResourceStore` contract in `XdsClient.java`, a `null` return
means "no subscription", while an empty collection means a **wildcard**
subscription — a real subscription that must still emit the empty
`resource_names` request and start its missing-resource timers.

The "DiscoveryRequest of this type has been sent on the current stream"
discriminator is tracked in a per-stream `sentTypes` set on `AdsStream`,
populated wherever a request is actually transmitted (initial sends,
ACKs, NACKs). Per-stream is the right scope because the server's view of
our subscriptions is per-stream — on stream replacement the set is
cleared implicitly along with the `AdsStream` instance.

Gating on `!versions.containsKey(type)` instead would be incorrect
because `versions` is only populated on ACK. If a watch is canceled
after the initial DiscoveryRequest goes out but before any response is
ACKed, that guard would suppress the empty unsubscribe — leaving the
server with a stale subscription until the stream resets. Tracking
actual sends per-stream closes that window.

The unsubscribe-all path (had-version, now-empty/null) is preserved: we
still send the empty request and clear the version, telling the server
to drop our subscription for that type.

## Mirrors grpc-go

This brings grpc-java in line with grpc-go's behavior. The Go equivalent
is
[`adsStreamImpl.sendExisting`](https://github.com/grpc/grpc-go/blob/v1.80.0/internal/xds/clients/xdsclient/ads_stream.go#L335-L368):

```go
for typ, state := range s.resourceTypeState {
    state.nonce = ""

    if len(state.subscribedResources) == 0 {
        continue                              // <-- explicit skip
    }

    names := resourceNames(state.subscribedResources)
    if err := s.sendMessageLocked(stream, names, typ.TypeURL, state.version, state.nonce, nil); err != nil {
        return err
    }
    s.startWatchTimersLocked(typ, names)
}
```

Two structural reasons grpc-go avoids this bug today:

1. The iteration domain is **per-stream**
([`s.resourceTypeState`](https://github.com/grpc/grpc-go/blob/v1.80.0/internal/xds/clients/xdsclient/ads_stream.go#L143)),
populated only when
[`subscribe`](https://github.com/grpc/grpc-go/blob/v1.80.0/internal/xds/clients/xdsclient/ads_stream.go#L167-L193)
is called for a resource on this stream — so a Cluster type never even
appears in the iteration of an EDS-only stream.
2. Even within that per-stream iteration, the explicit `if
len(state.subscribedResources) == 0 { continue }` covers the case where
the type has no subscription on this stream.

The grpc-java fix is the equivalent of (2). The
`!adsStream.sentTypes.contains` guard is needed because Java's iteration
domain is global (`getSubscribedResourceTypesWithTypeUrl` is
xds-client-wide), so we may see types we never subscribed to on this
stream.

Note that grpc-go physically separates two paths:
[`sendNewLocked`](https://github.com/grpc/grpc-go/blob/v1.80.0/internal/xds/clients/xdsclient/ads_stream.go#L308-L318)
handles runtime sub/unsub and sends every queued request unconditionally
(so an empty unsubscribe always goes on the wire, ACK or no ACK), while
`sendExisting` handles stream re-establishment and applies the `len ==
0` skip. grpc-java has a single `adjustResourceSubscription` function
that serves both paths — the per-stream `sentTypes` set is what lets the
same guard distinguish them: on a fresh stream `sentTypes` is empty so
the guard reduces to "no subscription → skip" (mirroring
`sendExisting`), while at runtime after the initial request
`sentTypes.contains(type)` is true so the guard does not trigger and the
empty unsubscribe is sent (mirroring `sendNewLocked`).

## Test plan

Unit tests in `ControlPlaneClientTest`:
- `streamReady_skipsEmptyDiscoveryRequestForUnsubscribedType` — the
federation case, asserts CDS request is suppressed and EDS still goes
through
- `streamReady_sendsRequestForAllTypesWhenAllHaveResources` — guards
against over-eager skip
- `streamReady_skipsTypeWithNoSubscription` — `null` return skips
- `streamReady_sendsWildcardRequestAndStartsTimers` — empty collection
(wildcard) still sends and starts timers
- `cancelBeforeAck_sendsEmptyUnsubscribe` — cancel-before-ACK timing
window still emits the unsubscribe

## Spec note

xDS SoTW spec ([Envoy xDS
protocol](https://www.envoyproxy.io/docs/envoy/latest/api-docs/xds_protocol))
treats an empty `resource_names` for LDS/CDS as a wildcard subscription
("send me everything of this type"). The previous grpc-java behavior
would unintentionally trigger wildcard CDS subscriptions on every
authority-specific stream — which an EDS-only server is right to refuse.
Skipping when no subscription exists side-steps that misinterpretation;
legitimate wildcard subscriptions (empty collection from a real
subscription) still go on the wire as intended, and the existing
unsubscribe-all path (with a prior version) continues to work.

Author: paulmurhy123

xds/src/main/java/io/grpc/xds/client/ControlPlaneClient.java

@@ -160,6 +160,31 @@ void adjustResourceSubscription(XdsResourceType<?> resourceType) {
     }
 
     Collection<String> resources = resourceStore.getSubscribedResources(serverInfo, resourceType);
+    if (resources == null && !adsStream.sentTypes.contains(resourceType)) {
+      // No subscription for this type on this server, and we have never sent a DiscoveryRequest
+      // of this type on the current stream — the server has no subscription state to clear.
+      //
+      // Per the ResourceStore contract in XdsClient.java, a null return means "no subscription";
+      // an empty collection means wildcard subscription, which is a real subscription and must
+      // not be skipped here.
+      //
+      // We track sent types per-stream rather than gating on `versions` because `versions` is
+      // only populated on ACK. If a watch is canceled after the initial DiscoveryRequest goes
+      // out but before any response is ACKed, `versions` would still have no entry for the
+      // type, and gating on it would suppress the empty unsubscribe — leaving the server with
+      // a stale subscription until the stream resets.
+      //
+      // Without this skip, sendDiscoveryRequests() iterates over every globally-subscribed
+      // resource type when a stream becomes ready and emits an empty DiscoveryRequest for types
+      // that have no subscription on this server. Per A47 (xDS Federation) servers may be
+      // authority-specific (e.g. an EDS-only control plane) and reject DiscoveryRequests for
+      // types they do not handle, tearing down the stream.
+      //
+      // Mirrors grpc-go's behavior in
+      // internal/xds/clients/xdsclient/ads_stream.go:sendExisting, which skips types with no
+      // subscription.
+      return;
+    }
     if (resources == null) {
       resources = Collections.emptyList();
     }
@@ -319,6 +344,11 @@ private class AdsStream implements XdsTransportFactory.EventHandler<DiscoveryRes
     // management server should not send a DiscoveryResponse for any DiscoveryRequest that has a
     // stale nonce."
     private final Map<String, String> respNonces = new HashMap<>();
+    // Resource types for which a DiscoveryRequest has been sent on this stream. Used by
+    // adjustResourceSubscription() to decide whether an empty unsubscribe must be sent on the
+    // wire: the server only has subscription state to clear for types we have actually sent a
+    // request for on this stream. Cleared implicitly when the stream is replaced.
+    private final Set<XdsResourceType<?>> sentTypes = new HashSet<>();
     private final StreamingCall<DiscoveryRequest, DiscoveryResponse> call;
     private final MethodDescriptor<DiscoveryRequest, DiscoveryResponse> methodDescriptor =
         AggregatedDiscoveryServiceGrpc.getStreamAggregatedResourcesMethod();
@@ -358,6 +388,7 @@ void sendDiscoveryRequest(XdsResourceType<?> type, String versionInfo,
       }
       DiscoveryRequest request = builder.build();
       call.sendMessage(request);
+      sentTypes.add(type);
       if (logger.isLoggable(XdsLogLevel.DEBUG)) {
         logger.log(XdsLogLevel.DEBUG, "Sent DiscoveryRequest\n{0}", messagePrinter.print(request));
       }

xds/src/test/java/io/grpc/xds/client/ControlPlaneClientTest.java

@@ -0,0 +1,279 @@
+/*
+ * Copyright 2026 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.xds.client;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.argThat;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.common.base.Stopwatch;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest;
+import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
+import io.grpc.InsecureChannelCredentials;
+import io.grpc.MethodDescriptor;
+import io.grpc.SynchronizationContext;
+import io.grpc.internal.BackoffPolicy;
+import io.grpc.internal.FakeClock;
+import io.grpc.xds.client.Bootstrapper.ServerInfo;
+import io.grpc.xds.client.EnvoyProtoData.Node;
+import io.grpc.xds.client.XdsClient.ResourceStore;
+import io.grpc.xds.client.XdsClient.XdsResponseHandler;
+import io.grpc.xds.client.XdsTransportFactory.EventHandler;
+import io.grpc.xds.client.XdsTransportFactory.StreamingCall;
+import io.grpc.xds.client.XdsTransportFactory.XdsTransport;
+import java.util.Collections;
+import java.util.Map;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnit;
+import org.mockito.junit.MockitoRule;
+
+/** Unit tests for {@link ControlPlaneClient}. */
+@RunWith(JUnit4.class)
+public class ControlPlaneClientTest {
+
+  private static final String CDS_TYPE_URL = "type.googleapis.com/envoy.config.cluster.v3.Cluster";
+  private static final String EDS_TYPE_URL =
+      "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment";
+
+  private final SynchronizationContext syncContext =
+      new SynchronizationContext((t, e) -> {
+        throw new AssertionError("Uncaught exception in sync context", e);
+      });
+  private final FakeClock fakeClock = new FakeClock();
+  private final ServerInfo serverInfo =
+      ServerInfo.create("eds-control-plane:8443", InsecureChannelCredentials.create());
+  private final Node bootstrapNode = Node.newBuilder().setId("test-node").build();
+
+  @Mock private XdsTransport xdsTransport;
+  @Mock private StreamingCall<DiscoveryRequest, DiscoveryResponse> streamingCall;
+  @Mock private XdsResponseHandler responseHandler;
+  @Mock private ResourceStore resourceStore;
+  @Mock private BackoffPolicy.Provider backoffPolicyProvider;
+  @Mock private MessagePrettyPrinter messagePrinter;
+  @Mock private XdsResourceType<?> cdsType;
+  @Mock private XdsResourceType<?> edsType;
+
+  @Rule public final MockitoRule mocks = MockitoJUnit.rule();
+
+  private ControlPlaneClient cpc;
+  private ArgumentCaptor<EventHandler<DiscoveryResponse>> handlerCaptor;
+
+  @Before
+  @SuppressWarnings("unchecked")
+  public void setUp() {
+    when(cdsType.typeUrl()).thenReturn(CDS_TYPE_URL);
+    when(cdsType.typeName()).thenReturn("CDS");
+    when(edsType.typeUrl()).thenReturn(EDS_TYPE_URL);
+    when(edsType.typeName()).thenReturn("EDS");
+
+    when(xdsTransport.<DiscoveryRequest, DiscoveryResponse>createStreamingCall(
+            anyString(),
+            any(MethodDescriptor.Marshaller.class),
+            any(MethodDescriptor.Marshaller.class)))
+        .thenReturn(streamingCall);
+    when(streamingCall.isReady()).thenReturn(true);
+
+    handlerCaptor = ArgumentCaptor.forClass(EventHandler.class);
+
+    cpc = new ControlPlaneClient(
+        xdsTransport,
+        serverInfo,
+        bootstrapNode,
+        responseHandler,
+        resourceStore,
+        fakeClock.getScheduledExecutorService(),
+        syncContext,
+        backoffPolicyProvider,
+        () -> Stopwatch.createUnstarted(fakeClock.getTicker()),
+        messagePrinter);
+  }
+
+  /**
+   * Reproduces the bug where, when an ADS stream is opened to an authority-specific server (e.g.
+   * an EDS-only control plane), {@code sendDiscoveryRequests} previously emitted an empty
+   * DiscoveryRequest for every globally-subscribed resource type — including types this server
+   * does not handle. Authority-specific servers may reject those requests with UNIMPLEMENTED and
+   * tear down the stream, blocking the legitimate request that follows.
+   *
+   * <p>Asserts that the empty CDS request is suppressed and only the EDS request (which has
+   * resources for this server) goes on the wire.
+   */
+  @Test
+  public void streamReady_skipsEmptyDiscoveryRequestForUnsubscribedType() {
+    // CDS is globally subscribed (e.g. against a different authority) but has no resources on
+    // this server. EDS has one resource on this server.
+    Map<String, XdsResourceType<?>> subscribedTypes =
+        ImmutableMap.of(CDS_TYPE_URL, cdsType, EDS_TYPE_URL, edsType);
+    when(resourceStore.getSubscribedResourceTypesWithTypeUrl()).thenReturn(subscribedTypes);
+    when(resourceStore.getSubscribedResources(serverInfo, cdsType)).thenReturn(null);
+    when(resourceStore.getSubscribedResources(serverInfo, edsType))
+        .thenReturn(ImmutableList.of("foo-endpoint"));
+
+    // Triggers stream creation and registers the EventHandler.
+    syncContext.execute(cpc::sendDiscoveryRequests);
+    verify(streamingCall).start(handlerCaptor.capture());
+
+    // Drive the stream into the connected state. onReady() flips sentInitialRequest=true and
+    // re-invokes sendDiscoveryRequests, which iterates the globally-subscribed types.
+    handlerCaptor.getValue().onReady();
+
+    // EDS request was sent with the one resource for this server.
+    ArgumentCaptor<DiscoveryRequest> sent = ArgumentCaptor.forClass(DiscoveryRequest.class);
+    verify(streamingCall, atLeastOnce()).sendMessage(sent.capture());
+    ImmutableSet<String> sentTypes = sent.getAllValues().stream()
+        .map(DiscoveryRequest::getTypeUrl)
+        .collect(ImmutableSet.toImmutableSet());
+    assertThat(sentTypes).contains(EDS_TYPE_URL);
+    assertThat(sentTypes).doesNotContain(CDS_TYPE_URL);
+
+    // Confirm the EDS request actually carried the resource name.
+    DiscoveryRequest edsReq = sent.getAllValues().stream()
+        .filter(r -> r.getTypeUrl().equals(EDS_TYPE_URL))
+        .findFirst()
+        .orElseThrow(() -> new AssertionError("EDS request not sent"));
+    assertThat(edsReq.getResourceNamesList()).containsExactly("foo-endpoint");
+  }
+
+  /**
+   * If a server has resources for every globally-subscribed type, the empty-skip guard is a
+   * no-op: a DiscoveryRequest is sent for every type. This guards against the skip becoming
+   * over-eager and dropping legitimate subscriptions.
+   */
+  @Test
+  public void streamReady_sendsRequestForAllTypesWhenAllHaveResources() {
+    Map<String, XdsResourceType<?>> subscribedTypes =
+        ImmutableMap.of(CDS_TYPE_URL, cdsType, EDS_TYPE_URL, edsType);
+    when(resourceStore.getSubscribedResourceTypesWithTypeUrl()).thenReturn(subscribedTypes);
+    when(resourceStore.getSubscribedResources(serverInfo, cdsType))
+        .thenReturn(ImmutableList.of("foo-cluster"));
+    when(resourceStore.getSubscribedResources(serverInfo, edsType))
+        .thenReturn(ImmutableList.of("foo-endpoint"));
+
+    syncContext.execute(cpc::sendDiscoveryRequests);
+    verify(streamingCall).start(handlerCaptor.capture());
+    handlerCaptor.getValue().onReady();
+
+    ArgumentCaptor<DiscoveryRequest> sent = ArgumentCaptor.forClass(DiscoveryRequest.class);
+    verify(streamingCall, times(2)).sendMessage(sent.capture());
+    ImmutableSet<String> sentTypes = sent.getAllValues().stream()
+        .map(DiscoveryRequest::getTypeUrl)
+        .collect(ImmutableSet.toImmutableSet());
+    assertThat(sentTypes).containsExactly(CDS_TYPE_URL, EDS_TYPE_URL);
+  }
+
+  /**
+   * If only one type has a subscription on this server, no request is sent for the unsubscribed
+   * type. This is the canonical multi-authority federation case (e.g. fabric authority owns CDS,
+   * eds-control-plane owns EDS — the eds-control-plane stream should only see EDS).
+   */
+  @Test
+  public void streamReady_skipsTypeWithNoSubscription() {
+    Map<String, XdsResourceType<?>> subscribedTypes =
+        ImmutableMap.of(CDS_TYPE_URL, cdsType, EDS_TYPE_URL, edsType);
+    when(resourceStore.getSubscribedResourceTypesWithTypeUrl()).thenReturn(subscribedTypes);
+    when(resourceStore.getSubscribedResources(serverInfo, cdsType)).thenReturn(null);
+    when(resourceStore.getSubscribedResources(serverInfo, edsType))
+        .thenReturn(ImmutableList.of("foo-endpoint"));
+
+    syncContext.execute(cpc::sendDiscoveryRequests);
+    verify(streamingCall).start(handlerCaptor.capture());
+    handlerCaptor.getValue().onReady();
+
+    verify(streamingCall, never()).sendMessage(
+        argThatTypeUrlIs(CDS_TYPE_URL));
+    verify(streamingCall).sendMessage(argThatTypeUrlIs(EDS_TYPE_URL));
+  }
+
+  /**
+   * Per the ResourceStore contract in XdsClient.java, an empty collection from
+   * getSubscribedResources indicates a wildcard subscription. The skip-on-empty guard must not
+   * suppress wildcard requests on initial stream ready — the server needs the empty-resource-list
+   * DiscoveryRequest to start streaming, and the watcher's missing-resource timers must start.
+   */
+  @Test
+  public void streamReady_sendsWildcardRequestAndStartsTimers() {
+    Map<String, XdsResourceType<?>> subscribedTypes = ImmutableMap.of(CDS_TYPE_URL, cdsType);
+    when(resourceStore.getSubscribedResourceTypesWithTypeUrl()).thenReturn(subscribedTypes);
+    // Empty collection == wildcard subscription per the ResourceStore contract.
+    when(resourceStore.getSubscribedResources(serverInfo, cdsType))
+        .thenReturn(Collections.emptyList());
+
+    syncContext.execute(cpc::sendDiscoveryRequests);
+    verify(streamingCall).start(handlerCaptor.capture());
+    handlerCaptor.getValue().onReady();
+
+    ArgumentCaptor<DiscoveryRequest> sent = ArgumentCaptor.forClass(DiscoveryRequest.class);
+    verify(streamingCall, atLeastOnce()).sendMessage(sent.capture());
+    DiscoveryRequest cdsReq = sent.getAllValues().stream()
+        .filter(r -> r.getTypeUrl().equals(CDS_TYPE_URL))
+        .findFirst()
+        .orElseThrow(() -> new AssertionError("CDS wildcard request not sent"));
+    assertThat(cdsReq.getResourceNamesList()).isEmpty();
+
+    verify(resourceStore).startMissingResourceTimers(Collections.emptyList(), cdsType);
+  }
+
+  /**
+   * If a watch is canceled after the initial DiscoveryRequest goes out but before any response
+   * is ACKed, the empty unsubscribe must still be sent — otherwise the server keeps the stale
+   * subscription until the stream resets. The skip guard must gate on per-stream send history,
+   * not on the {@code versions} map (which is only populated on ACK).
+   */
+  @Test
+  public void cancelBeforeAck_sendsEmptyUnsubscribe() {
+    Map<String, XdsResourceType<?>> subscribedTypes = ImmutableMap.of(CDS_TYPE_URL, cdsType);
+    when(resourceStore.getSubscribedResourceTypesWithTypeUrl()).thenReturn(subscribedTypes);
+    when(resourceStore.getSubscribedResources(serverInfo, cdsType))
+        .thenReturn(ImmutableList.of("foo-cluster"));
+
+    syncContext.execute(cpc::sendDiscoveryRequests);
+    verify(streamingCall).start(handlerCaptor.capture());
+    handlerCaptor.getValue().onReady();
+
+    // Initial DiscoveryRequest with the resource went out. No DiscoveryResponse has been ACKed.
+    verify(streamingCall).sendMessage(argThatTypeUrlIs(CDS_TYPE_URL));
+
+    // Cancel the watch before any response arrives: store now reports no subscription.
+    when(resourceStore.getSubscribedResources(serverInfo, cdsType)).thenReturn(null);
+    syncContext.execute(() -> cpc.adjustResourceSubscription(cdsType));
+
+    ArgumentCaptor<DiscoveryRequest> sent = ArgumentCaptor.forClass(DiscoveryRequest.class);
+    verify(streamingCall, times(2)).sendMessage(sent.capture());
+    DiscoveryRequest unsub = sent.getAllValues().get(1);
+    assertThat(unsub.getTypeUrl()).isEqualTo(CDS_TYPE_URL);
+    assertThat(unsub.getResourceNamesList()).isEmpty();
+  }
+
+  private static DiscoveryRequest argThatTypeUrlIs(String typeUrl) {
+    return argThat(req -> req != null && typeUrl.equals(req.getTypeUrl()));
+  }
+}