@Override
  public void expand(
      ExpansionApi.ExpansionRequest request,
      StreamObserver<ExpansionApi.ExpansionResponse> responseObserver) {
    if (!checkedAllServices) {
      try {
        waitForAllServicesToBeReady();
      } catch (TimeoutException e) {
        throw new RuntimeException(e);
      }
      checkedAllServices = true;
    }
    try {
      responseObserver.onNext(processExpand(request));
      responseObserver.onCompleted();
    } catch (RuntimeException exn) {
      responseObserver.onNext(
          ExpansionApi.ExpansionResponse.newBuilder()
              .setError(Throwables.getStackTraceAsString(exn))
              .build());
      responseObserver.onCompleted();
    }
  }
@Test
  public void testExpandFirstEndpoint() {
    ExpansionServiceClient expansionServiceClient = Mockito.mock(ExpansionServiceClient.class);
    Mockito.when(clientFactory.getExpansionServiceClient(Mockito.any()))
        .thenReturn(expansionServiceClient);
    Mockito.when(expansionServiceClient.expand(Mockito.any()))
        .thenReturn(
            ExpansionResponse.newBuilder()
                .setTransform(
                    PTransform.newBuilder()
                        .setSpec(FunctionSpec.newBuilder().setUrn("dummy_urn_1")))
                .build());

    ExpansionRequest request = ExpansionRequest.newBuilder().build();
    StreamObserver<ExpansionResponse> responseObserver = Mockito.mock(StreamObserver.class);
    expansionService.expand(request, responseObserver);
    Mockito.verify(expansionServiceClient, Mockito.times(1)).expand(request);

    ArgumentCaptor<ExpansionResponse> expansionResponseCapture =
        ArgumentCaptor.forClass(ExpansionResponse.class);
    Mockito.verify(responseObserver).onNext(expansionResponseCapture.capture());
    assertEquals(
        "dummy_urn_1", expansionResponseCapture.getValue().getTransform().getSpec().getUrn());
  }