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 testSetPythonExecutable() throws IOException {
        Configuration config = new Configuration();

        File zipFile = new File(tmpDirPath + File.separator + "venv.zip");
        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();
        }
        PythonEnvUtils.PythonEnvironment env = preparePythonEnvironment(config, null, tmpDirPath);
        if (OperatingSystem.isWindows()) {
            assertThat(env.pythonExec).isEqualTo("python.exe");
        } else {
            assertThat(env.pythonExec).isEqualTo("python");
        }

        Map<String, String> systemEnv = new HashMap<>(System.getenv());
        systemEnv.put(PYFLINK_CLIENT_EXECUTABLE, "python3");
        CommonTestUtils.setEnv(systemEnv);
        try {
            env = preparePythonEnvironment(config, null, tmpDirPath);
            assertThat(env.pythonExec).isEqualTo("python3");
        } finally {
            systemEnv.remove(PYFLINK_CLIENT_EXECUTABLE);
            CommonTestUtils.setEnv(systemEnv);
        }

        config.set(PYTHON_ARCHIVES, zipFile.getPath());
        systemEnv = new HashMap<>(System.getenv());
        systemEnv.put(PYFLINK_CLIENT_EXECUTABLE, "venv.zip/venv/bin/python");
        CommonTestUtils.setEnv(systemEnv);
        try {
            env = preparePythonEnvironment(config, null, tmpDirPath);
            assertThat(env.pythonExec).isEqualTo("venv.zip/venv/bin/python");
        } finally {
            systemEnv.remove(PYFLINK_CLIENT_EXECUTABLE);
            CommonTestUtils.setEnv(systemEnv);
        }
        java.nio.file.Path[] files =
                FileUtils.listDirectory(new File(env.archivesDirectory).toPath());
        assertThat(files).hasSize(1);
        assertThat(files[0].getFileName().toString()).isEqualTo(zipFile.getName());

        config.removeConfig(PYTHON_ARCHIVES);

        config.set(PYTHON_CLIENT_EXECUTABLE, "/usr/bin/python");
        env = preparePythonEnvironment(config, null, tmpDirPath);
        assertThat(env.pythonExec).isEqualTo("/usr/bin/python");
    }