Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/5396#discussion_r165343453
--- Diff:
flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStateOutputStreamTest.java
---
@@ -332,6 +335,43 @@ public void testMixedBelowAndAboveThreshold() throws
Exception {
assertTrue(isDirectoryEmpty(directory));
}
+ //
------------------------------------------------------------------------
+ // Not deleting parent directories
+ //
------------------------------------------------------------------------
+
+ /**
+ * This test checks that the stream does not check and clean the parent
directory
+ * when encountering a write error.
+ */
+ @Test
+ public void testStreamDoesNotTryToCleanUpParentOnError() throws
Exception {
+ final File directory = tempDir.newFolder();
+
+ // prevent creation of files in that directory
+ assertTrue(directory.setWritable(false, true));
+ checkDirectoryNotWritable(directory);
+
+ FileSystem fs = spy(FileSystem.getLocalFileSystem());
+
+ FsCheckpointStateOutputStream stream1 = new
FsCheckpointStateOutputStream(
+ Path.fromLocalFile(directory), fs, 1024, 1);
+
+ FsCheckpointStateOutputStream stream2 = new
FsCheckpointStateOutputStream(
+ Path.fromLocalFile(directory), fs, 1024, 1);
+
+ stream1.write(new byte[61]);
+ stream2.write(new byte[61]);
+
+ try {
+ stream1.closeAndGetHandle();
+ fail("this should fail with an exception");
+ } catch (IOException ignored) {}
+
+ stream2.close();
+
+ verify(fs, times(0)).delete(any(Path.class), anyBoolean());
--- End diff --
Will add an additional check that the directory still exists
---
