public static void main(String[] args) throws Exception {
    main(System::getenv);
  }
@Test
  @SuppressWarnings("FutureReturnValueIgnored") // failure will cause test to timeout.
  public void testLaunchFnHarnessAndTeardownCleanly() throws Exception {
    Function<String, String> environmentVariableMock = mock(Function.class);

    PipelineOptions options = PipelineOptionsFactory.create();

    when(environmentVariableMock.apply("HARNESS_ID")).thenReturn("id");
    when(environmentVariableMock.apply("PIPELINE_OPTIONS"))
        .thenReturn(PipelineOptionsTranslation.toJson(options));

    List<BeamFnApi.LogEntry> logEntries = new ArrayList<>();
    List<BeamFnApi.InstructionResponse> instructionResponses = mock(List.class);

    BeamFnLoggingGrpc.BeamFnLoggingImplBase loggingService =
        new BeamFnLoggingGrpc.BeamFnLoggingImplBase() {
          @Override
          public StreamObserver<BeamFnApi.LogEntry.List> logging(
              StreamObserver<LogControl> responseObserver) {
            return TestStreams.withOnNext(
                    (BeamFnApi.LogEntry.List entries) ->
                        logEntries.addAll(entries.getLogEntriesList()))
                .withOnCompleted(responseObserver::onCompleted)
                .build();
          }
        };

    BeamFnControlGrpc.BeamFnControlImplBase controlService =
        new BeamFnControlGrpc.BeamFnControlImplBase() {
          @Override
          public StreamObserver<InstructionResponse> control(
              StreamObserver<InstructionRequest> responseObserver) {
            CountDownLatch waitForResponses =
                new CountDownLatch(1 /* number of responses expected */);
            options
                .as(ExecutorOptions.class)
                .getScheduledExecutorService()
                .submit(
                    () -> {
                      responseObserver.onNext(INSTRUCTION_REQUEST);
                      Uninterruptibles.awaitUninterruptibly(waitForResponses);
                      responseObserver.onCompleted();
                    });
            return TestStreams.withOnNext(
                    (InstructionResponse t) -> {
                      instructionResponses.add(t);
                      waitForResponses.countDown();
                    })
                .withOnCompleted(waitForResponses::countDown)
                .build();
          }
        };

    Server loggingServer = ServerBuilder.forPort(0).addService(loggingService).build();
    loggingServer.start();
    try {
      Server controlServer = ServerBuilder.forPort(0).addService(controlService).build();
      controlServer.start();

      try {
        Endpoints.ApiServiceDescriptor loggingDescriptor =
            Endpoints.ApiServiceDescriptor.newBuilder()
                .setUrl("localhost:" + loggingServer.getPort())
                .build();
        Endpoints.ApiServiceDescriptor controlDescriptor =
            Endpoints.ApiServiceDescriptor.newBuilder()
                .setUrl("localhost:" + controlServer.getPort())
                .build();

        when(environmentVariableMock.apply("LOGGING_API_SERVICE_DESCRIPTOR"))
            .thenReturn(TextFormat.printer().printToString(loggingDescriptor));
        when(environmentVariableMock.apply("CONTROL_API_SERVICE_DESCRIPTOR"))
            .thenReturn(TextFormat.printer().printToString(controlDescriptor));

        FnHarness.main(environmentVariableMock);
      } finally {
        controlServer.shutdownNow();
      }
    } finally {
      loggingServer.shutdownNow();
    }

    // Verify that we first run onStartup functions before even reading the environment, and that
    // we then call beforeProcessing functions before executing instructions.
    InOrder inOrder =
        inOrder(onStartupMock, beforeProcessingMock, environmentVariableMock, instructionResponses);
    inOrder.verify(onStartupMock).run();
    inOrder.verify(environmentVariableMock, atLeastOnce()).apply(any());
    inOrder.verify(beforeProcessingMock).accept(any());
    inOrder.verify(instructionResponses).add(INSTRUCTION_RESPONSE);
  }