### Question:
V2DiscoveryServer extends DiscoveryServer<DiscoveryRequest, DiscoveryResponse> { public ClusterDiscoveryServiceGrpc.ClusterDiscoveryServiceImplBase getClusterDiscoveryServiceImpl() { return new ClusterDiscoveryServiceGrpc.ClusterDiscoveryServiceImplBase() { @Override public StreamObserver<DiscoveryRequest> streamClusters( StreamObserver<DiscoveryResponse> responseObserver) { return createRequestHandler(responseObserver, false, Resources.V2.CLUSTER_TYPE_URL); } }; } V2DiscoveryServer(ConfigWatcher configWatcher);  V2DiscoveryServer(DiscoveryServerCallbacks callbacks,
      ConfigWatcher configWatcher);  V2DiscoveryServer(
      List<DiscoveryServerCallbacks> callbacks,
      ConfigWatcher configWatcher);  V2DiscoveryServer(List<DiscoveryServerCallbacks> callbacks,
      ConfigWatcher configWatcher, ExecutorGroup executorGroup, ProtoResourcesSerializer protoResourcesSerializer); AggregatedDiscoveryServiceGrpc.AggregatedDiscoveryServiceImplBase getAggregatedDiscoveryServiceImpl(); ClusterDiscoveryServiceGrpc.ClusterDiscoveryServiceImplBase getClusterDiscoveryServiceImpl(); EndpointDiscoveryServiceGrpc.EndpointDiscoveryServiceImplBase getEndpointDiscoveryServiceImpl(); ListenerDiscoveryServiceGrpc.ListenerDiscoveryServiceImplBase getListenerDiscoveryServiceImpl(); RouteDiscoveryServiceGrpc.RouteDiscoveryServiceImplBase getRouteDiscoveryServiceImpl(); SecretDiscoveryServiceGrpc.SecretDiscoveryServiceImplBase getSecretDiscoveryServiceImpl();  }### Answer:
@Test public void testCallbacksOnCancelled() throws InterruptedException, ClassNotFoundException { final CountDownLatch streamCloseWithErrorLatch = new CountDownLatch(1); final CountDownLatch watchCreated = new CountDownLatch(1); AtomicReference<Watch> watch = new AtomicReference<>(); MockDiscoveryServerCallbacks callbacks = new MockDiscoveryServerCallbacks() { @Override public void onStreamCloseWithError(long streamId, String typeUrl, Throwable error) { assertThat(watch.get().isCancelled()).isTrue(); super.onStreamCloseWithError(streamId, typeUrl, error); streamCloseWithErrorLatch.countDown(); } }; MockConfigWatcher configWatcher = new MockConfigWatcher(false, createResponses()) { @Override public Watch createWatch(boolean ads, XdsRequest request, Set<String> knownResources, Consumer<Response> responseConsumer, boolean hasClusterChanged) { watchCreated.countDown(); watch.set(super.createWatch(ads, request, knownResources, responseConsumer, false)); return watch.get(); } }; V2DiscoveryServer server = new V2DiscoveryServer(callbacks, configWatcher); grpcServer.getServiceRegistry().addService(server.getClusterDiscoveryServiceImpl()); ClusterDiscoveryServiceStub stub = ClusterDiscoveryServiceGrpc.newStub(grpcServer.getChannel()); MockDiscoveryResponseObserver responseObserver = new MockDiscoveryResponseObserver(); StreamObserver<DiscoveryRequest> requestObserver = stub.streamClusters(responseObserver); requestObserver.onNext(DiscoveryRequest.newBuilder() .setNode(NODE) .setResponseNonce("1") .setTypeUrl(Resources.V2.CLUSTER_TYPE_URL) .setVersionInfo(VERSION) .build()); if (!watchCreated.await(1, TimeUnit.SECONDS)) { fail("failed to execute watchCreated callback before timeout"); } requestObserver.onError(Status.CANCELLED.asException()); if (!streamCloseWithErrorLatch.await(1, TimeUnit.SECONDS)) { fail("failed to execute onStreamCloseWithError callback before timeout"); } callbacks.assertThatNoErrors(); assertThat(callbacks.streamCloseCount).hasValue(0); assertThat(callbacks.streamCloseWithErrorCount).hasValue(1); assertThat(callbacks.streamOpenCount).hasValue(1); assertThat(callbacks.streamRequestCount).hasValue(1); assertThat(callbacks.streamResponseCount).hasValue(1); }