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
---