@Test public void actualFileSystem() throws IOException { final java.nio.file.Path tmpDir = Files.createTempDirectory("hdfs-reader-test-"); try { final Path rootPath = new Path(tmpDir.toString()); final FileSystem localFs = FileSystem.getLocal(new Configuration()); final FsBasedCheckpointer cptr = new FsBasedCheckpointer(localFs, (partitionId, instant) -> new Path(rootPath, partitionId.toString() + '.' + DateTimeFormatter.ofPattern("YYYY-MM-dd").format(instant.atZone(ZoneId.of("UTC"))))); final Instant firstDay = Instant.parse("2019-04-02T12:25:00.00Z"); final Instant secondDay = Instant.parse("2019-04-15T23:42:51.00Z"); localFs.mkdirs(rootPath); Assert.assertTrue(cptr.tryCheckpoint(1, firstDay)); Assert.assertFalse(cptr.tryCheckpoint(1, firstDay)); Assert.assertTrue(cptr.tryCheckpoint(2, firstDay)); Assert.assertTrue(cptr.tryCheckpoint(1, secondDay)); List<FileStatus> fileStatuses = new ArrayList<>(); Collections.addAll(fileStatuses, localFs.listStatus(rootPath)); Assert.assertEquals(3, fileStatuses.size()); Assert.assertTrue(fileStatuses.stream().anyMatch(f -> f.getPath().getName().equals("1.2019-04-02"))); Assert.assertTrue(fileStatuses.stream().anyMatch(f -> f.getPath().getName().equals("2.2019-04-02"))); Assert.assertTrue(fileStatuses.stream().anyMatch(f -> f.getPath().getName().equals("1.2019-04-15"))); } finally { FileUtils.deleteDirectory(tmpDir.toFile()); } }
@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);  }