@Test public void testDeleteOlderCleanMetaFiles() throws Exception { String[] instantTimes = new String[]{"20160501010101", "20160501020101", "20160501030101", "20160501040101", "20160502020601", "20160502030601", "20160502040601", "20160502050601", "20160506030611", "20160506040611", "20160506050611", "20160506060611"}; List<HoodieInstant> hoodieInstants = new ArrayList<>(); for (String instantTime : instantTimes) { Files.createFile(Paths.get(basePath, HoodieTableMetaClient.METAFOLDER_NAME, instantTime + HoodieTimeline.CLEAN_EXTENSION)); hoodieInstants.add(new HoodieInstant(false, HoodieTimeline.CLEAN_ACTION, instantTime)); } String metaPath = Paths.get(basePath, ".hoodie").toString(); FSUtils.deleteOlderCleanMetaFiles(FSUtils.getFs(basePath, new Configuration()), metaPath, hoodieInstants.stream()); File[] cleanFiles = new File(metaPath).listFiles((dir, name) -> name.contains(HoodieTimeline.CLEAN_EXTENSION)); assertNotNull(cleanFiles); assertEquals(cleanFiles.length, minCleanToKeep); }
public static void deleteOlderCleanMetaFiles(FileSystem fs, String metaPath, Stream<HoodieInstant> instants) { instants.skip(MIN_CLEAN_TO_KEEP).forEach(s -> { try { fs.delete(new Path(metaPath, s.getFileName()), false); } catch (IOException e) { throw new HoodieIOException("Could not delete clean meta files" + s.getFileName(), e); } }); }
FSUtils { public static void deleteOlderCleanMetaFiles(FileSystem fs, String metaPath, Stream<HoodieInstant> instants) { instants.skip(MIN_CLEAN_TO_KEEP).forEach(s -> { try { fs.delete(new Path(metaPath, s.getFileName()), false); } catch (IOException e) { throw new HoodieIOException("Could not delete clean meta files" + s.getFileName(), e); } }); } }
FSUtils { public static void deleteOlderCleanMetaFiles(FileSystem fs, String metaPath, Stream<HoodieInstant> instants) { instants.skip(MIN_CLEAN_TO_KEEP).forEach(s -> { try { fs.delete(new Path(metaPath, s.getFileName()), false); } catch (IOException e) { throw new HoodieIOException("Could not delete clean meta files" + s.getFileName(), e); } }); }  }
FSUtils { public static void deleteOlderCleanMetaFiles(FileSystem fs, String metaPath, Stream<HoodieInstant> instants) { instants.skip(MIN_CLEAN_TO_KEEP).forEach(s -> { try { fs.delete(new Path(metaPath, s.getFileName()), false); } catch (IOException e) { throw new HoodieIOException("Could not delete clean meta files" + s.getFileName(), e); } }); }  static Configuration prepareHadoopConf(Configuration conf); static FileSystem getFs(String path, Configuration conf); static FileSystem getFs(String path, Configuration conf, boolean localByDefault); static Path addSchemeIfLocalPath(String path); static String makeWriteToken(int taskPartitionId, int stageId, long taskAttemptId); static String makeDataFileName(String instantTime, String writeToken, String fileId); static String makeDataFileName(String instantTime, String writeToken, String fileId, String fileExtension); static String makeBootstrapIndexFileName(String instantTime, String fileId, String ext); static String maskWithoutFileId(String instantTime, int taskPartitionId); static String getCommitFromCommitFile(String commitFileName); static String getCommitTime(String fullFileName); static long getFileSize(FileSystem fs, Path path); static String getFileId(String fullFileName); static List<String> getAllPartitionFoldersThreeLevelsDown(FileSystem fs, String basePath); static String getRelativePartitionPath(Path basePath, Path fullPartitionPath); static List<String> getAllFoldersWithPartitionMetaFile(FileSystem fs, String basePathStr); static void processFiles(FileSystem fs, String basePathStr, Function<FileStatus, Boolean> consumer,
                                  boolean excludeMetaFolder); static List<String> getAllPartitionPaths(FileSystem fs, String basePathStr, boolean assumeDatePartitioning); static String getFileExtension(String fullName); static String createNewFileIdPfx(); static String getFileExtensionFromLog(Path logPath); static String getFileIdFromLogPath(Path path); static String getFileIdFromFilePath(Path filePath); static String getBaseCommitTimeFromLogPath(Path path); static Integer getTaskPartitionIdFromLogPath(Path path); static String getWriteTokenFromLogPath(Path path); static Integer getStageIdFromLogPath(Path path); static Integer getTaskAttemptIdFromLogPath(Path path); static int getFileVersionFromLog(Path logPath); static String makeLogFileName(String fileId, String logFileExtension, String baseCommitTime, int version,
      String writeToken); static boolean isLogFile(Path logPath); static Option<HoodieLogFile> getLatestLogFile(Stream<HoodieLogFile> logFiles); static Stream<HoodieLogFile> getAllLogFiles(FileSystem fs, Path partitionPath, final String fileId,
      final String logFileExtension, final String baseCommitTime); static Option<Pair<Integer, String>> getLatestLogVersion(FileSystem fs, Path partitionPath,
      final String fileId, final String logFileExtension, final String baseCommitTime); static int computeNextLogVersion(FileSystem fs, Path partitionPath, final String fileId,
      final String logFileExtension, final String baseCommitTime); static int getDefaultBufferSize(final FileSystem fs); static Short getDefaultReplication(FileSystem fs, Path path); static boolean recoverDFSFileLease(final DistributedFileSystem dfs, final Path p); static void deleteOlderCleanMetaFiles(FileSystem fs, String metaPath, Stream<HoodieInstant> instants); static void deleteOlderRollbackMetaFiles(FileSystem fs, String metaPath, Stream<HoodieInstant> instants); static void deleteInstantFile(FileSystem fs, String metaPath, HoodieInstant instant); static void createPathIfNotExists(FileSystem fs, Path partitionPath); static Long getSizeInMB(long sizeInBytes); static Path getPartitionPath(String basePath, String partitionPath); static Path getPartitionPath(Path basePath, String partitionPath); static String getDFSFullPartitionPath(FileSystem fs, Path fullPartitionPath); static boolean isGCSInputStream(FSDataInputStream inputStream); static Configuration registerFileSystem(Path file, Configuration conf); static HoodieWrapperFileSystem getFs(String path, SerializableConfiguration hadoopConf,
      ConsistencyGuardConfig consistencyGuardConfig); static List<FileStatus> getGlobStatusExcludingMetaFolder(FileSystem fs, Path globPath); }
FSUtils { public static void deleteOlderCleanMetaFiles(FileSystem fs, String metaPath, Stream<HoodieInstant> instants) { instants.skip(MIN_CLEAN_TO_KEEP).forEach(s -> { try { fs.delete(new Path(metaPath, s.getFileName()), false); } catch (IOException e) { throw new HoodieIOException("Could not delete clean meta files" + s.getFileName(), e); } }); }  static Configuration prepareHadoopConf(Configuration conf); static FileSystem getFs(String path, Configuration conf); static FileSystem getFs(String path, Configuration conf, boolean localByDefault); static Path addSchemeIfLocalPath(String path); static String makeWriteToken(int taskPartitionId, int stageId, long taskAttemptId); static String makeDataFileName(String instantTime, String writeToken, String fileId); static String makeDataFileName(String instantTime, String writeToken, String fileId, String fileExtension); static String makeBootstrapIndexFileName(String instantTime, String fileId, String ext); static String maskWithoutFileId(String instantTime, int taskPartitionId); static String getCommitFromCommitFile(String commitFileName); static String getCommitTime(String fullFileName); static long getFileSize(FileSystem fs, Path path); static String getFileId(String fullFileName); static List<String> getAllPartitionFoldersThreeLevelsDown(FileSystem fs, String basePath); static String getRelativePartitionPath(Path basePath, Path fullPartitionPath); static List<String> getAllFoldersWithPartitionMetaFile(FileSystem fs, String basePathStr); static void processFiles(FileSystem fs, String basePathStr, Function<FileStatus, Boolean> consumer,
                                  boolean excludeMetaFolder); static List<String> getAllPartitionPaths(FileSystem fs, String basePathStr, boolean assumeDatePartitioning); static String getFileExtension(String fullName); static String createNewFileIdPfx(); static String getFileExtensionFromLog(Path logPath); static String getFileIdFromLogPath(Path path); static String getFileIdFromFilePath(Path filePath); static String getBaseCommitTimeFromLogPath(Path path); static Integer getTaskPartitionIdFromLogPath(Path path); static String getWriteTokenFromLogPath(Path path); static Integer getStageIdFromLogPath(Path path); static Integer getTaskAttemptIdFromLogPath(Path path); static int getFileVersionFromLog(Path logPath); static String makeLogFileName(String fileId, String logFileExtension, String baseCommitTime, int version,
      String writeToken); static boolean isLogFile(Path logPath); static Option<HoodieLogFile> getLatestLogFile(Stream<HoodieLogFile> logFiles); static Stream<HoodieLogFile> getAllLogFiles(FileSystem fs, Path partitionPath, final String fileId,
      final String logFileExtension, final String baseCommitTime); static Option<Pair<Integer, String>> getLatestLogVersion(FileSystem fs, Path partitionPath,
      final String fileId, final String logFileExtension, final String baseCommitTime); static int computeNextLogVersion(FileSystem fs, Path partitionPath, final String fileId,
      final String logFileExtension, final String baseCommitTime); static int getDefaultBufferSize(final FileSystem fs); static Short getDefaultReplication(FileSystem fs, Path path); static boolean recoverDFSFileLease(final DistributedFileSystem dfs, final Path p); static void deleteOlderCleanMetaFiles(FileSystem fs, String metaPath, Stream<HoodieInstant> instants); static void deleteOlderRollbackMetaFiles(FileSystem fs, String metaPath, Stream<HoodieInstant> instants); static void deleteInstantFile(FileSystem fs, String metaPath, HoodieInstant instant); static void createPathIfNotExists(FileSystem fs, Path partitionPath); static Long getSizeInMB(long sizeInBytes); static Path getPartitionPath(String basePath, String partitionPath); static Path getPartitionPath(Path basePath, String partitionPath); static String getDFSFullPartitionPath(FileSystem fs, Path fullPartitionPath); static boolean isGCSInputStream(FSDataInputStream inputStream); static Configuration registerFileSystem(Path file, Configuration conf); static HoodieWrapperFileSystem getFs(String path, SerializableConfiguration hadoopConf,
      ConsistencyGuardConfig consistencyGuardConfig); static List<FileStatus> getGlobStatusExcludingMetaFolder(FileSystem fs, Path globPath);  }