public class PythonDependencyManager {

  // create PythonDependencyManager from ExecutionConfig.getGlobalJobParameters().toMap() and

  // distributedCaches.

  public static PythonDependencyManager create(

     Map<String, String> dependencyMetaData,

     DistributedCache distributedCache) {...}

  // key is the absolute path of the files to append to PYTHONPATH, value is the origin file name

  public Map<String, String> getPythonFiles() {...} 

  // absolute path of requirements.txt

  public String getRequirementsFilePath() {...}

  // absolute path of the cached directory which contains user provided python packages

  public String getRequirementsDirPath() {...}

  //path of the python executable file

  public String getPythonExec() {...}

  // key is the name of the environment variable, value is the value of the environment variable

  public Map<String, String> getEnvironmentVariable() {...}

  // key is the absolute path of the zip file, value is the target directory name to be extracted to

  public Map<String, String> getArchives() {...}

}

public interface PythonEnvironmentManager {

  /**

   * Create Apache Beam Environment object of python worker.

   */

  RunnerApi.Environment createEnvironment();

  /**

   * Create the RetrievalToken file which records all the files that need to be transferred via Apache Beam's 

   * ArtifactService.

   */

  String createRetrievalToken();

  /**

   * Delete generated files during above actions.

   */

  void cleanup();

}

public class ProcessEnvironmentManager implements PythonEnvironmentManager {

  public static ProcessEnvironmentManager create(

     PythonDependencyManager dependencyManager,

     String tmpDirectoryBase,

     Map<String, String> systemEnv) {

   }

  public ProcessEnvironmentManager(...) {

    prepareEnvironment();

  }

  @Override

  public void cleanup() {

     // perform the clean up work

     removeShutdownHook();

  }

  @Override

  public RunnerApi.Environment createEnvironment() {

     // command = path of udf runner

     return Environments.createProcessEnvironment("", "", command, generateEnvironmentVariable());

  }

  @Override

  public String createRetrievalToken() {

      // File transfer is unnecessary in process mode,

      // just create an empty RetrievalToken.

      return emptyRetrievalToken;

  }

  private Map<String, String> generateEnvironmentVariable() {

      // construct the environment variables such as PYTHONPATH, etc

  }

  private void prepareEnvironment() {

     registerShutdownHook();

     prepareWorkingDir();

  }

  private void prepareWorkingDir() {...}

  private Thread registerShutdownHook() {

     Thread thread = new Thread(new DeleteTemporaryFilesHook(pythonTmpDirectory));

     Runtime.getRuntime().addShutdownHook(thread);

     return thread;

  }

}

public class DockerEnvironmentManager implements PythonEnvironmentManager {

  public static DockerEnvironmentManager create(

     PythonDependencyManager dependencyManager,

     String tmpDirectoryBase,

     String dockerImageUrl) {

  }

  public DockerEnvironmentManager(...) {

    registerShutdownHook();

  }

  @Override

  public void cleanup() {

     // perform the clean up work

     removeShutdownHook();

  }

  @Override

  public RunnerApi.Environment createEnvironment() {

     return Environments.createDockerEnvironment(dockerImageUrl);

  }

  @Override

  public String createRetrievalToken() {

      // construct the RetrievalToken according to user uploaded files

  }

  private Thread registerShutdownHook() {

     Thread thread = new Thread(new DeleteTemporaryFilesHook(pythonTmpDirectory));

     Runtime.getRuntime().addShutdownHook(thread);

     return thread;

  }

}