@Test public void testDeleteApplicationFiles() throws Exception { final Path applicationFilesDir = temporaryFolder.newFolder(".flink").toPath(); Files.createFile(applicationFilesDir.resolve("flink.jar")); try (Stream<Path> files = Files.list(temporaryFolder.getRoot().toPath())) { assertThat(files.count(), equalTo(1L)); } try (Stream<Path> files = Files.list(applicationFilesDir)) { assertThat(files.count(), equalTo(1L)); } Utils.deleteApplicationFiles(Collections.singletonMap( YarnConfigKeys.FLINK_YARN_FILES, applicationFilesDir.toString())); try (Stream<Path> files = Files.list(temporaryFolder.getRoot().toPath())) { assertThat(files.count(), equalTo(0L)); } }
public static void deleteApplicationFiles(final Map<String, String> env) { final String applicationFilesDir = env.get(YarnConfigKeys.FLINK_YARN_FILES); if (!StringUtils.isNullOrWhitespaceOnly(applicationFilesDir)) { final org.apache.flink.core.fs.Path path = new org.apache.flink.core.fs.Path(applicationFilesDir); try { final org.apache.flink.core.fs.FileSystem fileSystem = path.getFileSystem(); if (!fileSystem.delete(path, true)) { LOG.error("Deleting yarn application files under {} was unsuccessful.", applicationFilesDir); } } catch (final IOException e) { LOG.error("Could not properly delete yarn application files directory {}.", applicationFilesDir, e); } } else { LOG.debug("No yarn application files directory set. Therefore, cannot clean up the data."); } }
Utils { public static void deleteApplicationFiles(final Map<String, String> env) { final String applicationFilesDir = env.get(YarnConfigKeys.FLINK_YARN_FILES); if (!StringUtils.isNullOrWhitespaceOnly(applicationFilesDir)) { final org.apache.flink.core.fs.Path path = new org.apache.flink.core.fs.Path(applicationFilesDir); try { final org.apache.flink.core.fs.FileSystem fileSystem = path.getFileSystem(); if (!fileSystem.delete(path, true)) { LOG.error("Deleting yarn application files under {} was unsuccessful.", applicationFilesDir); } } catch (final IOException e) { LOG.error("Could not properly delete yarn application files directory {}.", applicationFilesDir, e); } } else { LOG.debug("No yarn application files directory set. Therefore, cannot clean up the data."); } } }
Utils { public static void deleteApplicationFiles(final Map<String, String> env) { final String applicationFilesDir = env.get(YarnConfigKeys.FLINK_YARN_FILES); if (!StringUtils.isNullOrWhitespaceOnly(applicationFilesDir)) { final org.apache.flink.core.fs.Path path = new org.apache.flink.core.fs.Path(applicationFilesDir); try { final org.apache.flink.core.fs.FileSystem fileSystem = path.getFileSystem(); if (!fileSystem.delete(path, true)) { LOG.error("Deleting yarn application files under {} was unsuccessful.", applicationFilesDir); } } catch (final IOException e) { LOG.error("Could not properly delete yarn application files directory {}.", applicationFilesDir, e); } } else { LOG.debug("No yarn application files directory set. Therefore, cannot clean up the data."); } } private  Utils(); }
Utils { public static void deleteApplicationFiles(final Map<String, String> env) { final String applicationFilesDir = env.get(YarnConfigKeys.FLINK_YARN_FILES); if (!StringUtils.isNullOrWhitespaceOnly(applicationFilesDir)) { final org.apache.flink.core.fs.Path path = new org.apache.flink.core.fs.Path(applicationFilesDir); try { final org.apache.flink.core.fs.FileSystem fileSystem = path.getFileSystem(); if (!fileSystem.delete(path, true)) { LOG.error("Deleting yarn application files under {} was unsuccessful.", applicationFilesDir); } } catch (final IOException e) { LOG.error("Could not properly delete yarn application files directory {}.", applicationFilesDir, e); } } else { LOG.debug("No yarn application files directory set. Therefore, cannot clean up the data."); } } private  Utils(); static int calculateHeapSize(int memory, org.apache.flink.configuration.Configuration conf); static void setupYarnClassPath(Configuration conf, Map<String, String> appMasterEnv); static void deleteApplicationFiles(final Map<String, String> env); static void setTokensFor(ContainerLaunchContext amContainer, List<Path> paths, Configuration conf); static void addToEnvironment(Map<String, String> environment,
			String variable, String value); static Map<String, String> getEnvironmentVariables(String envPrefix, org.apache.flink.configuration.Configuration flinkConfiguration); }
Utils { public static void deleteApplicationFiles(final Map<String, String> env) { final String applicationFilesDir = env.get(YarnConfigKeys.FLINK_YARN_FILES); if (!StringUtils.isNullOrWhitespaceOnly(applicationFilesDir)) { final org.apache.flink.core.fs.Path path = new org.apache.flink.core.fs.Path(applicationFilesDir); try { final org.apache.flink.core.fs.FileSystem fileSystem = path.getFileSystem(); if (!fileSystem.delete(path, true)) { LOG.error("Deleting yarn application files under {} was unsuccessful.", applicationFilesDir); } } catch (final IOException e) { LOG.error("Could not properly delete yarn application files directory {}.", applicationFilesDir, e); } } else { LOG.debug("No yarn application files directory set. Therefore, cannot clean up the data."); } } private  Utils(); static int calculateHeapSize(int memory, org.apache.flink.configuration.Configuration conf); static void setupYarnClassPath(Configuration conf, Map<String, String> appMasterEnv); static void deleteApplicationFiles(final Map<String, String> env); static void setTokensFor(ContainerLaunchContext amContainer, List<Path> paths, Configuration conf); static void addToEnvironment(Map<String, String> environment,
			String variable, String value); static Map<String, String> getEnvironmentVariables(String envPrefix, org.apache.flink.configuration.Configuration flinkConfiguration); static final String KEYTAB_FILE_NAME; static final String KRB5_FILE_NAME; static final String YARN_SITE_FILE_NAME; static final int REMOTE_RESOURCES_FETCH_NUM_RETRY; static final int REMOTE_RESOURCES_FETCH_WAIT_IN_MILLI; }