@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); }
@Override public boolean tryCheckpoint(int partitionId, Instant when) { Path path = checkpointPathGenerator.apply(partitionId, when); try { if (!checkpointed(path)) { checkpoint(path); checkpointsCache.refresh(path); return true; } } catch (IOException | ExecutionException e) { LOGGER.warn("Couldn't write checkpoint file: {}", e.getMessage()); } return false; }
FsBasedCheckpointer implements Checkpointer { @Override public boolean tryCheckpoint(int partitionId, Instant when) { Path path = checkpointPathGenerator.apply(partitionId, when); try { if (!checkpointed(path)) { checkpoint(path); checkpointsCache.refresh(path); return true; } } catch (IOException | ExecutionException e) { LOGGER.warn("Couldn't write checkpoint file: {}", e.getMessage()); } return false; } }
FsBasedCheckpointer implements Checkpointer { @Override public boolean tryCheckpoint(int partitionId, Instant when) { Path path = checkpointPathGenerator.apply(partitionId, when); try { if (!checkpointed(path)) { checkpoint(path); checkpointsCache.refresh(path); return true; } } catch (IOException | ExecutionException e) { LOGGER.warn("Couldn't write checkpoint file: {}", e.getMessage()); } return false; } FsBasedCheckpointer(FileSystem fs, BiFunction<Integer, Instant, Path> checkpointPathGenerator); }
FsBasedCheckpointer implements Checkpointer { @Override public boolean tryCheckpoint(int partitionId, Instant when) { Path path = checkpointPathGenerator.apply(partitionId, when); try { if (!checkpointed(path)) { checkpoint(path); checkpointsCache.refresh(path); return true; } } catch (IOException | ExecutionException e) { LOGGER.warn("Couldn't write checkpoint file: {}", e.getMessage()); } return false; } FsBasedCheckpointer(FileSystem fs, BiFunction<Integer, Instant, Path> checkpointPathGenerator); @Override boolean tryCheckpoint(int partitionId, Instant when); }
FsBasedCheckpointer implements Checkpointer { @Override public boolean tryCheckpoint(int partitionId, Instant when) { Path path = checkpointPathGenerator.apply(partitionId, when); try { if (!checkpointed(path)) { checkpoint(path); checkpointsCache.refresh(path); return true; } } catch (IOException | ExecutionException e) { LOGGER.warn("Couldn't write checkpoint file: {}", e.getMessage()); } return false; } FsBasedCheckpointer(FileSystem fs, BiFunction<Integer, Instant, Path> checkpointPathGenerator); @Override boolean tryCheckpoint(int partitionId, Instant when);  }