@Test
public void createNonExistingCheckpoints() throws IOException {

    FileSystem fsMock = mock(FileSystem.class);
    Instant firstDateTime = TestUtils.instantFromDate("1987-08-13 12:12:00");
    Instant secondDateTime = TestUtils.instantFromDate("2019-04-02 12:25:00");
    Path firstPath = new Path("1.1987-08-13");
    Path secondPath = new Path("2.2019-04-02");
    FSDataOutputStream firstOsMock = mock(FSDataOutputStream.class);
    FSDataOutputStream secondOsMock = mock(FSDataOutputStream.class);
    FsBasedCheckpointer cptr = new FsBasedCheckpointer(fsMock, (partition, instant) ->
        new Path(partition.toString() + "." +
                DateTimeFormatter.ofPattern("YYYY-MM-dd").format(instant.atZone(ZoneId.of("UTC")))));
    when(fsMock.exists(firstPath)).thenReturn(false);
    when(fsMock.exists(secondPath)).thenReturn(false);
    when(fsMock.create(firstPath)).thenReturn(firstOsMock);
    when(fsMock.create(secondPath)).thenReturn(secondOsMock);
    Assert.assertTrue(cptr.tryCheckpoint(1, firstDateTime));
    verify(fsMock, times(2)).exists(firstPath);
    verify(fsMock, times(1)).create(firstPath);
    Assert.assertTrue(cptr.tryCheckpoint(2, secondDateTime));
    verify(fsMock, times(2)).exists(secondPath);
    verify(fsMock, times(1)).create(secondPath);
    verifyNoMoreInteractions(fsMock);
    verify(firstOsMock, times(1)).close();
    verify(secondOsMock, times(1)).close();
    verifyNoMoreInteractions(firstOsMock);
}