[FLINK-17721][filesystems][test] Fix test instability for AbstractHadoopFileSystemITTest.

flink-core/src/test/java/org/apache/flink/core/fs/FileSystemTestUtils.java

@@ -29,16 +29,17 @@ public class FileSystemTestUtils {
 
 	/**
 	 * Verifies that the given path eventually appears on / disappears from <tt>fs</tt> within
-	 * <tt>deadline</tt> nanoseconds.
+	 * <tt>consistencyToleranceNS</tt> nanoseconds.
 	 */
 	public static void checkPathEventualExistence(
 			FileSystem fs,
 			Path path,
 			boolean expectedExists,
-			long deadline) throws IOException, InterruptedException {
+			long consistencyToleranceNS) throws IOException, InterruptedException {
 		boolean dirExists;
+		long deadline = System.nanoTime() + consistencyToleranceNS;
 		while ((dirExists = fs.exists(path)) != expectedExists &&
-				System.nanoTime() < deadline) {
+				System.nanoTime() - deadline < 0) {
 			Thread.sleep(10);
 		}
 		assertEquals(expectedExists, dirExists);

flink-filesystems/flink-hadoop-fs/src/test/java/org/apache/flink/runtime/fs/hdfs/AbstractHadoopFileSystemITTest.java

@@ -49,22 +49,22 @@ public abstract class AbstractHadoopFileSystemITTest extends TestLogger {
 
 	protected static FileSystem fs;
 	protected static Path basePath;
-	protected static long deadline;
+	protected static long consistencyToleranceNS;
 
 	public static void checkPathExistence(Path path,
 			boolean expectedExists,
-			long deadline) throws IOException, InterruptedException {
-		if (deadline == 0) {
+			long consistencyToleranceNS) throws IOException, InterruptedException {
+		if (consistencyToleranceNS == 0) {
 			//strongly consistency
 			assertEquals(expectedExists, fs.exists(path));
 		} else {
 			//eventually consistency
-			checkPathEventualExistence(fs, path, expectedExists, deadline);
+			checkPathEventualExistence(fs, path, expectedExists, consistencyToleranceNS);
 		}
 	}
 
 	protected void checkEmptyDirectory(Path path) throws IOException, InterruptedException {
-		checkPathExistence(path, true, deadline);
+		checkPathExistence(path, true, consistencyToleranceNS);
 	}
 
 	@Test
@@ -80,7 +80,7 @@ public void testSimpleFileWriteAndRead() throws Exception {
 			}
 
 			// just in case, wait for the path to exist
-			checkPathExistence(path, true, deadline);
+			checkPathExistence(path, true, consistencyToleranceNS);
 
 			try (FSDataInputStream in = fs.open(path);
 				InputStreamReader ir = new InputStreamReader(in, StandardCharsets.UTF_8);
@@ -93,7 +93,7 @@ public void testSimpleFileWriteAndRead() throws Exception {
 			fs.delete(path, false);
 		}
 
-		checkPathExistence(path, false, deadline);
+		checkPathExistence(path, false, consistencyToleranceNS);
 	}
 
 	@Test
@@ -122,7 +122,7 @@ public void testDirectoryListing() throws Exception {
 				}
 				// just in case, wait for the file to exist (should then also be reflected in the
 				// directory's file list below)
-				checkPathExistence(file, true, deadline);
+				checkPathExistence(file, true, consistencyToleranceNS);
 			}
 
 			FileStatus[] files = fs.listStatus(directory);
@@ -138,24 +138,26 @@ public void testDirectoryListing() throws Exception {
 		}
 		finally {
 			// clean up
-			cleanupDirectoryWithRetry(fs, directory, deadline);
+			cleanupDirectoryWithRetry(fs, directory, consistencyToleranceNS);
 		}
 	}
 
 	@AfterClass
 	public static void teardown() throws IOException, InterruptedException {
 		try {
 			if (fs != null) {
-				cleanupDirectoryWithRetry(fs, basePath, deadline);
+				cleanupDirectoryWithRetry(fs, basePath, consistencyToleranceNS);
 			}
 		}
 		finally {
 			FileSystem.initialize(new Configuration());
 		}
 	}
 
-	public static void cleanupDirectoryWithRetry(FileSystem fs, Path path, long deadline) throws IOException, InterruptedException {
-		while (fs.exists(path) && System.nanoTime() < deadline) {
+	public static void cleanupDirectoryWithRetry(FileSystem fs, Path path, long consistencyToleranceNS) throws IOException, InterruptedException {
+		fs.delete(path, true);
+		long deadline = System.nanoTime() + consistencyToleranceNS;
+		while (fs.exists(path) && System.nanoTime() - deadline < 0) {
 			fs.delete(path, true);
 			Thread.sleep(50L);
 		}

flink-filesystems/flink-oss-fs-hadoop/src/test/java/org/apache/flink/fs/osshadoop/HadoopOSSFileSystemITCase.java

@@ -51,7 +51,7 @@ public static void setup() throws IOException {
 		FileSystem.initialize(conf);
 		basePath = new Path(OSSTestCredentials.getTestBucketUri() + TEST_DATA_DIR);
 		fs = basePath.getFileSystem();
-		deadline = 0;
+		consistencyToleranceNS = 0;
 	}
 
 	@Test

flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/HadoopS3FileSystemITCase.java

@@ -53,7 +53,7 @@ public static void setup() throws IOException {
 
 		basePath = new Path(S3TestCredentials.getTestBucketUri() + "tests-" + UUID.randomUUID());
 		fs = basePath.getFileSystem();
-		deadline = System.nanoTime() + 90_000_000_000L;
+		consistencyToleranceNS = 30_000_000_000L; // 30 seconds
 
 		// check for uniqueness of the test directory
 		// directory must not yet exist

flink-filesystems/flink-s3-fs-presto/src/test/java/org/apache/flink/fs/s3presto/PrestoS3FileSystemITCase.java

@@ -69,7 +69,7 @@ public static void setup() throws IOException {
 
 		basePath = new Path(S3TestCredentials.getTestBucketUri() + TEST_DATA_DIR);
 		fs = basePath.getFileSystem();
-		deadline = System.nanoTime() + 90_000_000_000L;
+		consistencyToleranceNS = 30_000_000_000L; // 30 seconds
 
 		// check for uniqueness of the test directory
 		// directory must not yet exist