@Override
  public void startWorker(
      StartWorkerRequest request, StreamObserver<StartWorkerResponse> responseObserver) {
    LOG.info(
        "Starting worker {} pointing at {}.",
        request.getWorkerId(),
        request.getControlEndpoint().getUrl());
    LOG.debug("Worker request {}.", request);

    Endpoints.ApiServiceDescriptor loggingEndpoint = request.getLoggingEndpoint();
    Endpoints.ApiServiceDescriptor controlEndpoint = request.getControlEndpoint();
    Set<String> runnerCapabilites = Collections.emptySet();
    if (request.hasProvisionEndpoint()) {
      ManagedChannelFactory channelFactory =
          ManagedChannelFactory.createDefault()
              .withInterceptors(
                  ImmutableList.of(AddHarnessIdInterceptor.create(request.getWorkerId())));

      ProvisionServiceGrpc.ProvisionServiceBlockingStub provisionStub =
          ProvisionServiceGrpc.newBlockingStub(
              channelFactory.forDescriptor(request.getProvisionEndpoint()));
      ProvisionApi.ProvisionInfo provisionInfo =
          provisionStub
              .getProvisionInfo(ProvisionApi.GetProvisionInfoRequest.newBuilder().build())
              .getInfo();

      runnerCapabilites = Sets.newHashSet(provisionInfo.getRunnerCapabilitiesList());
      if (provisionInfo.hasControlEndpoint()) {
        controlEndpoint = provisionInfo.getControlEndpoint();
      }
      if (provisionInfo.hasLoggingEndpoint()) {
        loggingEndpoint = provisionInfo.getLoggingEndpoint();
      }
    }
    // Lambda closured variables must be final.
    final Endpoints.ApiServiceDescriptor logEndpoint = loggingEndpoint;
    final Endpoints.ApiServiceDescriptor ctrlEndpoint = controlEndpoint;
    final Set<String> capabilities = runnerCapabilites;
    Thread th =
        new Thread(
            () -> {
              try {
                FnHarness.main(
                    request.getWorkerId(), options, capabilities, logEndpoint, ctrlEndpoint, null);
                LOG.info("Successfully started worker {}.", request.getWorkerId());
              } catch (Exception exn) {
                LOG.error(String.format("Failed to start worker %s.", request.getWorkerId()), exn);
              }
            });
    th.setName("SDK-worker-" + request.getWorkerId());
    th.setDaemon(true);
    th.start();

    responseObserver.onNext(StartWorkerResponse.newBuilder().build());
    responseObserver.onCompleted();
  }
@Test
  public void startWorker() {
    PipelineOptions options = PipelineOptionsFactory.create();
    StartWorkerRequest request = StartWorkerRequest.getDefaultInstance();
    StreamObserver<StartWorkerResponse> responseObserver = mock(StreamObserver.class);
    ExternalWorkerService service = new ExternalWorkerService(options);
    service.startWorker(request, responseObserver);

    verify(responseObserver).onNext(any(StartWorkerResponse.class));
    verify(responseObserver).onCompleted();
  }