@Override
  public RemoteEnvironment createEnvironment(Environment environment, String workerId)
      throws Exception {
    Preconditions.checkState(
        environment
            .getUrn()
            .equals(BeamUrns.getUrn(RunnerApi.StandardEnvironments.Environments.PROCESS)),
        "The passed environment does not contain a ProcessPayload.");
    final RunnerApi.ProcessPayload processPayload =
        RunnerApi.ProcessPayload.parseFrom(environment.getPayload());

    String executable = processPayload.getCommand();
    String provisionEndpoint = provisioningServiceServer.getApiServiceDescriptor().getUrl();

    String semiPersistDir = pipelineOptions.as(RemoteEnvironmentOptions.class).getSemiPersistDir();
    ImmutableList.Builder<String> argsBuilder =
        ImmutableList.<String>builder()
            .add(String.format("--id=%s", workerId))
            .add(String.format("--provision_endpoint=%s", provisionEndpoint));
    if (semiPersistDir != null) {
      argsBuilder.add(String.format("--semi_persist_dir=%s", semiPersistDir));
    }

    LOG.debug("Creating Process for worker ID {}", workerId);
    // Wrap the blocking call to clientSource.get in case an exception is thrown.
    InstructionRequestHandler instructionHandler = null;
    try {
      ProcessManager.RunningProcess process =
          processManager.startProcess(
              workerId, executable, argsBuilder.build(), processPayload.getEnvMap());
      // Wait on a client from the gRPC server.
      while (instructionHandler == null) {
        try {
          // If the process is not alive anymore, we abort.
          process.isAliveOrThrow();
          instructionHandler = clientSource.take(workerId, Duration.ofSeconds(5));
        } catch (TimeoutException timeoutEx) {
          LOG.info(
              "Still waiting for startup of environment '{}' for worker id {}",
              processPayload.getCommand(),
              workerId);
        } catch (InterruptedException interruptEx) {
          Thread.currentThread().interrupt();
          throw new RuntimeException(interruptEx);
        }
      }
    } catch (Exception e) {
      try {
        processManager.stopProcess(workerId);
      } catch (Exception processKillException) {
        e.addSuppressed(processKillException);
      }
      throw e;
    }

    return ProcessEnvironment.create(processManager, environment, workerId, instructionHandler);
  }
@Test
  public void destroysCorrectContainer() throws Exception {
    RemoteEnvironment handle = factory.createEnvironment(ENVIRONMENT, "workerId");
    handle.close();
    verify(processManager).stopProcess("workerId");
  }