tillrohrmann commented on a change in pull request #18083: URL: https://github.com/apache/flink/pull/18083#discussion_r785021549
########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypointUtils.java ########## @@ -128,4 +134,106 @@ public static void configureUncaughtExceptionHandler(Configuration config) { new ClusterUncaughtExceptionHandler( config.get(ClusterOptions.UNCAUGHT_EXCEPTION_HANDLING))); } + + /** + * Creates the working directory for the TaskManager process. This method ensures that the + * working directory exists. + * + * @param configuration to extract the required settings from + * @param resourceId identifying the TaskManager process + * @return working directory + * @throws IOException if the working directory could not be created + */ + public static WorkingDirectory createTaskManagerWorkingDirectory( + Configuration configuration, ResourceID resourceId) throws IOException { + return WorkingDirectory.create( + generateTaskManagerWorkingDirectoryFile(configuration, resourceId)); + } + + /** + * Generates the working directory {@link File} for the TaskManager process. This method does + * not ensure that the working directory exists. + * + * @param configuration to extract the required settings from + * @param resourceId identifying the TaskManager process + * @return working directory file + */ + @VisibleForTesting + public static File generateTaskManagerWorkingDirectoryFile( + Configuration configuration, ResourceID resourceId) { + return generateWorkingDirectoryFile( + configuration, + Optional.of(ClusterOptions.TASK_MANAGER_PROCESS_WORKING_DIR_BASE), + "tm_" + resourceId); + } + + /** + * Generates the working directory {@link File} for the JobManager process. This method does not + * ensure that the working directory exists. + * + * @param configuration to extract the required settings from + * @param resourceId identifying the JobManager process + * @return working directory file + */ + @VisibleForTesting + public static File generateJobManagerWorkingDirectoryFile( + Configuration configuration, ResourceID resourceId) { + return generateWorkingDirectoryFile( + configuration, + Optional.of(ClusterOptions.JOB_MANAGER_PROCESS_WORKING_DIR_BASE), + "jm_" + resourceId); + } + + /** + * Generate the working directory from the given configuration. If a preceding option is + * specified, then this config option will be read first for the working directory. Next {@link + * ClusterOptions#PROCESS_WORKING_DIR_BASE} will be tried. At last, {@link CoreOptions#TMP_DIRS} + * will be used to extract the working directory base from. + * + * @param configuration to extract the working directory from + * @param precedingOption optional preceding option + * @param workingDirectoryName name of the working directory to create + * @return working directory + */ + public static File generateWorkingDirectoryFile( + Configuration configuration, + Optional<ConfigOption<String>> precedingOption, + String workingDirectoryName) { + final Optional<String> optionalWorkingDirectory = + getOptionalWorkingDirectory(configuration, precedingOption); + + final File workingDirectoryBase = + optionalWorkingDirectory + .map(File::new) + .orElseGet(() -> ConfigurationUtils.getFirstTempDirectory(configuration)); + + return new File(workingDirectoryBase, workingDirectoryName); + } + + private static Optional<String> getOptionalWorkingDirectory( + Configuration configuration, Optional<ConfigOption<String>> precedingOption) { Review comment: Agreed. Will change it. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org