static PythonEnvironment preparePythonEnvironment(
            ReadableConfig config, String entryPointScript, String tmpDir) throws IOException {
        PythonEnvironment env = new PythonEnvironment();

        // 1. set the path of python interpreter.
        String pythonExec =
                config.getOptional(PYTHON_CLIENT_EXECUTABLE)
                        .orElse(System.getenv(PYFLINK_CLIENT_EXECUTABLE));
        if (pythonExec != null) {
            env.pythonExec = pythonExec;
        }

        // 2. setup temporary local directory for the user files
        tmpDir = new File(tmpDir).getAbsolutePath();
        Path tmpDirPath = new Path(tmpDir);
        tmpDirPath.getFileSystem().mkdirs(tmpDirPath);
        env.tempDirectory = tmpDir;

        // 3. append the internal lib files to PYTHONPATH.
        if (System.getenv(ConfigConstants.ENV_FLINK_OPT_DIR) != null) {
            String pythonLibDir =
                    System.getenv(ConfigConstants.ENV_FLINK_OPT_DIR) + File.separator + "python";
            env.pythonPath =
                    getLibFiles(pythonLibDir).stream()
                            .map(p -> p.toFile().getAbsolutePath())
                            .collect(Collectors.joining(File.pathSeparator));
        }

        // 4. copy relevant python files to tmp dir and set them in PYTHONPATH.
        if (config.getOptional(PYTHON_FILES).isPresent()) {
            List<Path> pythonFiles =
                    Arrays.stream(config.get(PYTHON_FILES).split(FILE_DELIMITER))
                            .map(Path::new)
                            .collect(Collectors.toList());
            addToPythonPath(env, pythonFiles);
        }

        // 5. set the archives directory as the working directory, then user could access the
        // content of the archives via relative path
        if (config.getOptional(PYTHON_ARCHIVES).isPresent()
                && (config.getOptional(PYTHON_CLIENT_EXECUTABLE).isPresent()
                        || !StringUtils.isNullOrWhitespaceOnly(
                                System.getenv(PYFLINK_CLIENT_EXECUTABLE)))) {
            env.archivesDirectory = String.join(File.separator, tmpDir, PYTHON_ARCHIVES_DIR);

            // extract archives to archives directory
            config.getOptional(PYTHON_ARCHIVES)
                    .ifPresent(
                            pyArchives -> {
                                for (String archive : pyArchives.split(FILE_DELIMITER)) {
                                    final Path archivePath;
                                    final String targetDirName;
                                    final String originalFileName;
                                    if (archive.contains(PythonDependencyUtils.PARAM_DELIMITER)) {
                                        String[] filePathAndTargetDir =
                                                archive.split(
                                                        PythonDependencyUtils.PARAM_DELIMITER, 2);
                                        archivePath = new Path(filePathAndTargetDir[0]);
                                        targetDirName = filePathAndTargetDir[1];
                                        originalFileName = archivePath.getName();
                                    } else {
                                        archivePath = new Path(archive);
                                        originalFileName = archivePath.getName();
                                        targetDirName = originalFileName;
                                    }

                                    Path localArchivePath = archivePath;
                                    try {
                                        if (archivePath.getFileSystem().isDistributedFS()) {
                                            localArchivePath =
                                                    new Path(
                                                            env.tempDirectory,
                                                            String.join(
                                                                    File.separator,
                                                                    UUID.randomUUID().toString(),
                                                                    originalFileName));
                                            FileUtils.copy(archivePath, localArchivePath, false);
                                        }
                                    } catch (IOException e) {
                                        String msg =
                                                String.format(
                                                        "Error occurred when copying %s to %s.",
                                                        archivePath, localArchivePath);
                                        throw new RuntimeException(msg, e);
                                    }

                                    try {
                                        CompressionUtils.extractFile(
                                                localArchivePath.getPath(),
                                                String.join(
                                                        File.separator,
                                                        env.archivesDirectory,
                                                        targetDirName),
                                                originalFileName);
                                    } catch (IOException e) {
                                        throw new RuntimeException(
                                                "Extract archives to archives directory failed.",
                                                e);
                                    }
                                }
                            });
        }

        // 4. append configured python.pythonpath to the PYTHONPATH.
        if (config.getOptional(PYTHON_PATH).isPresent()) {
            env.pythonPath =
                    String.join(
                            File.pathSeparator,
                            config.getOptional(PYTHON_PATH).get(),
                            env.pythonPath);
        }

        if (entryPointScript != null) {
            addToPythonPath(env, Collections.singletonList(new Path(entryPointScript)));
        }
        return env;
    }
@Test
    void testPreparePythonEnvironment() throws IOException {
        // Skip this test on Windows as we can not control the Window Driver letters.
        assumeThat(OperatingSystem.isWindows()).isFalse();

        // xxx/a.zip, xxx/subdir/b.py, xxx/subdir/c.zip
        File zipFile = new File(tmpDirPath + File.separator + "a.zip");
        File dirFile = new File(tmpDirPath + File.separator + "module_dir");
        File subdirFile = new File(tmpDirPath + File.separator + "subdir");
        File relativeFile =
                new File(tmpDirPath + File.separator + "subdir" + File.separator + "b.py");
        File schemeFile =
                new File(tmpDirPath + File.separator + "subdir" + File.separator + "c.egg");

        // The files must actually exist
        try (ZipArchiveOutputStream zipOut =
                new ZipArchiveOutputStream(new FileOutputStream(zipFile))) {
            ZipArchiveEntry entry = new ZipArchiveEntry("zipDir" + "/zipfile0");
            zipOut.putArchiveEntry(entry);
            zipOut.write(new byte[] {1, 1, 1, 1, 1});
            zipOut.closeArchiveEntry();
        }
        dirFile.mkdir();
        subdirFile.mkdir();
        relativeFile.createNewFile();
        schemeFile.createNewFile();

        String workingDir = new File("").getAbsolutePath();
        String absolutePath = relativeFile.getAbsolutePath();

        Path zipPath = new Path(zipFile.getAbsolutePath());
        Path dirPath = new Path(dirFile.getAbsolutePath());
        Path relativePath =
                new Path(Paths.get(workingDir).relativize(Paths.get(absolutePath)).toString());
        Path schemePath = new Path("file://" + schemeFile.getAbsolutePath());

        List<Path> pyFilesList = new ArrayList<>();
        pyFilesList.add(zipPath);
        pyFilesList.add(dirPath);
        pyFilesList.add(relativePath);
        pyFilesList.add(schemePath);

        String pyFiles =
                pyFilesList.stream()
                        .map(Path::toString)
                        .collect(Collectors.joining(FILE_DELIMITER));

        Configuration config = new Configuration();
        config.set(PYTHON_FILES, pyFiles);

        PythonEnvUtils.PythonEnvironment env = preparePythonEnvironment(config, null, tmpDirPath);

        String base = replaceUUID(env.tempDirectory);
        Set<String> expectedPythonPaths = new HashSet<>();
        expectedPythonPaths.add(String.join(File.separator, base, "{uuid}", "a"));
        expectedPythonPaths.add(String.join(File.separator, base, "{uuid}", "module_dir"));
        expectedPythonPaths.add(String.join(File.separator, base, "{uuid}"));
        expectedPythonPaths.add(String.join(File.separator, base, "{uuid}", "c.egg"));

        Set<String> actualPaths =
                Arrays.stream(env.pythonPath.split(File.pathSeparator))
                        .map(PythonEnvUtilsTest::replaceUUID)
                        .collect(Collectors.toSet());
        assertThat(actualPaths).isEqualTo(expectedPythonPaths);
    }