ProcessEnvironmentFactory implements EnvironmentFactory { @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); InstructionRequestHandler instructionHandler = null; try { ProcessManager.RunningProcess process = processManager.startProcess( workerId, executable, argsBuilder.build(), processPayload.getEnvMap()); while (instructionHandler == null) { try { 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); } private  ProcessEnvironmentFactory(
      ProcessManager processManager,
      GrpcFnServer<StaticGrpcProvisionService> provisioningServiceServer,
      IdGenerator idGenerator,
      ControlClientPool.Source clientSource,
      PipelineOptions pipelineOptions); static ProcessEnvironmentFactory create(
      ProcessManager processManager,
      GrpcFnServer<StaticGrpcProvisionService> provisioningServiceServer,
      ControlClientPool.Source clientSource,
      IdGenerator idGenerator,
      PipelineOptions pipelineOptions); @Override RemoteEnvironment createEnvironment(Environment environment, String workerId);  }
@Test public void createsCorrectEnvironment() throws Exception { RemoteEnvironment handle = factory.createEnvironment(ENVIRONMENT, "workerId"); assertThat(handle.getInstructionRequestHandler(), is(client)); assertThat(handle.getEnvironment(), equalTo(ENVIRONMENT)); Mockito.verify(processManager).startProcess(eq("workerId"), anyString(), anyList(), anyMap()); }
@Test public void destroysCorrectContainer() throws Exception { RemoteEnvironment handle = factory.createEnvironment(ENVIRONMENT, "workerId"); handle.close(); verify(processManager).stopProcess("workerId"); }
@Test public void createsMultipleEnvironments() throws Exception { Environment fooEnv = Environments.createProcessEnvironment("", "", "foo", Collections.emptyMap()); RemoteEnvironment fooHandle = factory.createEnvironment(fooEnv, "workerId"); assertThat(fooHandle.getEnvironment(), is(equalTo(fooEnv))); Environment barEnv = Environments.createProcessEnvironment("", "", "bar", Collections.emptyMap()); RemoteEnvironment barHandle = factory.createEnvironment(barEnv, "workerId"); assertThat(barHandle.getEnvironment(), is(equalTo(barEnv))); }