[FLINK-5345] [core] Add a utility to delete directories without failing in the...

[FLINK-5345] [core] Add a utility to delete directories without failing in the presence of concurrent deletes

flink-core/src/main/java/org/apache/flink/util/FileUtils.java

@@ -23,27 +23,28 @@ import org.apache.flink.core.fs.FileSystem;
 import org.apache.flink.core.fs.Path;
 
 import java.io.File;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
 import java.nio.file.StandardOpenOption;
 
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
 /**
- * This is a utility class to deal with temporary files.
+ * This is a utility class to deal files and directories. Contains utilities for recursive
+ * deletion and creation of temporary files.
  */
 public final class FileUtils {
 
-	/**
-	 * The alphabet to construct the random part of the filename from.
-	 */
-	private static final char[] ALPHABET = { '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', '0', 'a', 'b', 'c', 'd',
-		'e', 'f' };
+	/** The alphabet to construct the random part of the filename from. */
+	private static final char[] ALPHABET = 
+			{ '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', '0', 'a', 'b', 'c', 'd', 'e', 'f' };
 
-	/**
-	 * The length of the random part of the filename.
-	 */
-	private static final int LENGTH = 12;
+	/** The length of the random part of the filename. */
+	private static final int RANDOM_FILE_NAME_LENGTH = 12;
 
-	
+	// ------------------------------------------------------------------------
 
 	/**
 	 * Constructs a random filename with the given prefix and
@@ -54,10 +55,9 @@ public final class FileUtils {
 	 * @return the generated random filename with the given prefix
 	 */
 	public static String getRandomFilename(final String prefix) {
-
 		final StringBuilder stringBuilder = new StringBuilder(prefix);
 
-		for (int i = 0; i < LENGTH; i++) {
+		for (int i = 0; i < RANDOM_FILE_NAME_LENGTH; i++) {
 			stringBuilder.append(ALPHABET[(int) Math.floor(Math.random() * (double) ALPHABET.length)]);
 		}
 
@@ -87,7 +87,150 @@ public final class FileUtils {
 	}
 
 	// ------------------------------------------------------------------------
-	//  Deleting directories
+	//  Deleting directories on standard File Systems
+	// ------------------------------------------------------------------------
+
+	/**
+	 * Removes the given file or directory recursively.
+	 * 
+	 * <p>If the file or directory does not exist, this does not throw an exception, but simply does nothing.
+	 * It considers the fact that a file-to-be-deleted is not present a success.
+	 * 
+	 * <p>This method is safe against other concurrent deletion attempts.
+	 * 
+	 * @param file The file or directory to delete.
+	 * 
+	 * @throws IOException Thrown if the directory could not be cleaned for some reason, for example
+	 *                     due to missing access/write permissions.
+	 */
+	public static void deleteFileOrDirectory(File file) throws IOException {
+		checkNotNull(file, "file");
+
+		if (file.isDirectory()) {
+			// file exists and is directory
+			deleteDirectory(file);
+		}
+		else if (file.exists()) {
+			try {
+				Files.delete(file.toPath());
+			}
+			catch (NoSuchFileException e) {
+				// if the file is already gone (concurrently), we don't mind
+			}
+		}
+		// else: already deleted
+	}
+
+	/**
+	 * Deletes the given directory recursively.
+	 * 
+	 * <p>If the directory does not exist, this does not throw an exception, but simply does nothing.
+	 * It considers the fact that a directory-to-be-deleted is not present a success.
+	 * 
+	 * <p>This method is safe against other concurrent deletion attempts.
+	 * 
+	 * @param directory The directory to be deleted.
+	 * @throws IOException Thrown if the given file is not a directory, or if the directory could not be
+	 *                     deleted for some reason, for example due to missing access/write permissions.
+	 */
+	public static void deleteDirectory(File directory) throws IOException {
+		checkNotNull(directory, "directory");
+
+		if (directory.isDirectory()) {
+			// directory exists and is a directory
+
+			// empty the directory first
+			try {
+				cleanDirectory(directory);
+			}
+			catch (FileNotFoundException ignored) {
+				// someone concurrently deleted the directory, nothing to do for us
+				return;
+			}
+
+			// delete the directory. this fails if the directory is not empty, meaning
+			// if new files got concurrently created. we want to fail then.
+			try {
+				Files.delete(directory.toPath());
+			}
+			catch (NoSuchFileException ignored) {
+				// if someone else deleted this concurrently, we don't mind
+				// the result is the same for us, after all
+			}
+		}
+		else if (directory.exists()) {
+			// exists but is file, not directory
+			// either an error from the caller, or concurrently a file got created
+			throw new IOException(directory + " is not a directory");
+		}
+		// else: does not exist, which is okay (as if deleted)
+	}
+
+	/**
+	 * Deletes the given directory recursively, not reporting any I/O exceptions
+	 * that occur.
+	 * 
+	 * <p>This method is identical to {@link FileUtils#deleteDirectory(File)}, except that it
+	 * swallows all exceptions and may leave the job quietly incomplete.
+	 * 
+	 * @param directory The directory to delete.
+	 */
+	public static void deleteDirectoryQuietly(File directory) {
+		if (directory == null) {
+			return;
+		}
+
+		// delete and do not report if it fails
+		try {
+			deleteDirectory(directory);
+		} catch (Exception ignored) {}
+	}
+
+	/**
+	 * Removes all files contained within a directory, without removing the directory itself. 
+	 * 
+	 * <p>This method is safe against other concurrent deletion attempts.
+	 * 
+	 * @param directory The directory to remove all files from.
+	 * 
+	 * @throws FileNotFoundException Thrown if the directory itself does not exist.
+	 * @throws IOException Thrown if the file indicates a proper file and not a directory, or if
+	 *                     the directory could not be cleaned for some reason, for example
+	 *                     due to missing access/write permissions.
+	 */
+	public static void cleanDirectory(File directory) throws IOException, FileNotFoundException {
+		checkNotNull(directory, "directory");
+
+		if (directory.isDirectory()) {
+			final File[] files = directory.listFiles();
+
+			if (files == null) {
+				// directory does not exist any more or no permissions
+				if (directory.exists()) {
+					throw new IOException("Failed to list contents of " + directory);
+				} else {
+					throw new FileNotFoundException(directory.toString());
+				}
+			}
+
+			// remove all files in the directory
+			for (File file : files) {
+				if (file != null) {
+					deleteFileOrDirectory(file);
+				}
+			}
+		}
+		else if (directory.exists()) {
+			throw new IOException(directory + " is not a directory but a regular file");
+		}
+		else {
+			// else does not exist at all
+			throw new FileNotFoundException(directory.toString());
+		}
+	}
+
+	// ------------------------------------------------------------------------
+	//  Deleting directories on Flink FileSystem abstraction
 	// ------------------------------------------------------------------------
 
 	/**

flink-core/src/test/java/org/apache/flink/util/FileUtilsTest.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.util;
+
+import org.apache.flink.core.testutils.CheckedThread;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.file.AccessDeniedException;
+import java.nio.file.Files;
+import java.util.Random;
+
+import static org.junit.Assert.*;
+import static org.junit.Assume.*;
+
+public class FileUtilsTest {
+
+	@Rule
+	public final TemporaryFolder tmp = new TemporaryFolder();
+
+	// ------------------------------------------------------------------------
+	//  Tests
+	// ------------------------------------------------------------------------
+
+	@Test
+	public void testDeleteQuietly() throws Exception {
+		// should ignore the call
+		FileUtils.deleteDirectoryQuietly(null);
+
+		File doesNotExist = new File(tmp.getRoot(), "abc");
+		FileUtils.deleteDirectoryQuietly(doesNotExist);
+
+		File cannotDeleteParent = tmp.newFolder();
+		File cannotDeleteChild = new File(cannotDeleteParent, "child");
+
+		try {
+			assumeTrue(cannotDeleteChild.createNewFile());
+			assumeTrue(cannotDeleteParent.setWritable(false));
+			assumeTrue(cannotDeleteChild.setWritable(false));
+
+			FileUtils.deleteDirectoryQuietly(cannotDeleteParent);
+		}
+		finally {
+			//noinspection ResultOfMethodCallIgnored
+			cannotDeleteParent.setWritable(true);
+			//noinspection ResultOfMethodCallIgnored
+			cannotDeleteChild.setWritable(true);
+		}
+	}
+
+	@Test
+	public void testDeleteDirectory() throws Exception {
+
+		// deleting a non-existent file should not cause an error
+
+		File doesNotExist = new File(tmp.newFolder(), "abc");
+		FileUtils.deleteDirectory(doesNotExist);
+
+		// deleting a write protected file should throw an error
+
+		File cannotDeleteParent = tmp.newFolder();
+		File cannotDeleteChild = new File(cannotDeleteParent, "child");
+
+		try {
+			assumeTrue(cannotDeleteChild.createNewFile());
+			assumeTrue(cannotDeleteParent.setWritable(false));
+			assumeTrue(cannotDeleteChild.setWritable(false));
+
+			FileUtils.deleteDirectory(cannotDeleteParent);
+			fail("this should fail with an exception");
+		}
+		catch (AccessDeniedException ignored) {
+			// this is expected
+		}
+		finally {
+			//noinspection ResultOfMethodCallIgnored
+			cannotDeleteParent.setWritable(true);
+			//noinspection ResultOfMethodCallIgnored
+			cannotDeleteChild.setWritable(true);
+		}
+	}
+
+	@Test
+	public void testDeleteDirectoryConcurrently() throws Exception {
+		final File parent = tmp.newFolder();
+
+		generateRandomDirs(parent, 20, 5, 3);
+
+		// start three concurrent threads that delete the contents
+		CheckedThread t1 = new Deleter(parent);
+		CheckedThread t2 = new Deleter(parent);
+		CheckedThread t3 = new Deleter(parent);
+		t1.start();
+		t2.start();
+		t3.start();
+		t1.sync();
+		t2.sync();
+		t3.sync();
+
+		// assert is empty
+		assertFalse(parent.exists());
+	}
+
+	// ------------------------------------------------------------------------
+	//  Utilities
+	// ------------------------------------------------------------------------
+
+	private static void generateRandomDirs(File dir, int numFiles, int numDirs, int depth) throws IOException {
+		// generate the random files
+		for (int i = 0; i < numFiles; i++) {
+			File file = new File(dir, new AbstractID().toString());
+			try (FileOutputStream out = new FileOutputStream(file)) {
+				out.write(1);
+			}
+		}
+
+		if (depth > 0) {
+			// generate the directories
+			for (int i = 0; i < numDirs; i++) {
+				File subdir = new File(dir, new AbstractID().toString());
+				assertTrue(subdir.mkdir());
+				generateRandomDirs(subdir, numFiles, numDirs, depth - 1);
+			}
+		}
+	}
+
+	// ------------------------------------------------------------------------
+
+	private static class Deleter extends CheckedThread {
+
+		private final File target;
+
+		Deleter(File target) {
+			this.target = target;
+		}
+
+		@Override
+		public void go() throws Exception {
+			FileUtils.deleteDirectory(target);
+		}
+	}
+}