@Test public void recreateExistingFsCheckpoint() throws IOException { FileSystem fsMock = mock(FileSystem.class); Path returnedPath = new Path("testPath"); when(fsMock.exists(returnedPath)).thenReturn(true); FsBasedCheckpointer cptr = new FsBasedCheckpointer(fsMock, (ignored, alsoIgnored) -> returnedPath); Assert.assertFalse(cptr.tryCheckpoint(1, Instant.MIN)); FsBasedCheckpointer otherCptr = new FsBasedCheckpointer(fsMock, (ignored, alsoIgnored) -> returnedPath); Assert.assertFalse(otherCptr.tryCheckpoint(1, Instant.MIN)); verify(fsMock, times(2)).exists(returnedPath); }
@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);  }