MetadataCleanupJob { public Map<String, Long> cleanup(boolean delete, int jobOutdatedDays) throws Exception { CubeManager cubeManager = CubeManager.getInstance(config); long newResourceTimeCut = System.currentTimeMillis() - NEW_RESOURCE_THREADSHOLD_MS; FileSystem fs = HadoopUtil.getWorkingFileSystem(HadoopUtil.getCurrentConfiguration()); Map<String, Long> toDeleteCandidates = Maps.newHashMap(); for (String resourceRoot : new String[] { ResourceStore.SNAPSHOT_RESOURCE_ROOT, ResourceStore.CUBE_STATISTICS_ROOT, ResourceStore.EXT_SNAPSHOT_RESOURCE_ROOT }) { for (String dir : noNull(store.listResources(resourceRoot))) { for (String res : noNull(store.listResources(dir))) { long timestamp = getTimestamp(res); if (timestamp < newResourceTimeCut) toDeleteCandidates.put(res, timestamp); } } } try { FileStatus[] fStatus = new FileStatus[0]; fStatus = ArrayUtils.addAll(fStatus, fs.listStatus(new Path( KylinConfig.getInstanceFromEnv().getHdfsWorkingDirectory() + "resources/GlobalDict/dict"))); fStatus = ArrayUtils.addAll(fStatus, fs.listStatus(new Path( KylinConfig.getInstanceFromEnv().getHdfsWorkingDirectory() + "resources/SegmentDict/dict"))); for (FileStatus status : fStatus) { String path = status.getPath().toString(); FileStatus[] globalDicts = fs.listStatus(new Path(path)); for (FileStatus globalDict : globalDicts) { String globalDictPath = globalDict.getPath().toString(); long timestamp = getTimestamp(globalDict); if (timestamp < newResourceTimeCut) toDeleteCandidates.put(globalDictPath, timestamp); } } } catch (FileNotFoundException e) { logger.info("Working Directory does not exist on HDFS. "); } for (String resourceRoot : new String[] { ResourceStore.DICT_RESOURCE_ROOT }) { for (String dir : noNull(store.listResources(resourceRoot))) { for (String dir2 : noNull(store.listResources(dir))) { for (String res : noNull(store.listResources(dir2))) { long timestamp = getTimestamp(res); if (timestamp < newResourceTimeCut) toDeleteCandidates.put(res, timestamp); } } } } Set<String> activeResources = Sets.newHashSet(); for (CubeInstance cube : cubeManager.reloadAndListAllCubes()) { activeResources.addAll(cube.getSnapshots().values()); for (CubeSegment segment : cube.getSegments()) { activeResources.addAll(segment.getSnapshotPaths()); activeResources.addAll(segment.getDictionaryPaths()); activeResources.add(segment.getStatisticsResourcePath()); for (String dictPath : segment.getDictionaryPaths()) { DictionaryInfo dictInfo = store.getResource(dictPath, DictionaryInfoSerializer.FULL_SERIALIZER); if ("org.apache.kylin.dict.AppendTrieDictionary" .equals(dictInfo != null ? dictInfo.getDictionaryClass() : null)) { String dictObj = dictInfo.getDictionaryObject().toString(); String basedir = dictObj.substring(dictObj.indexOf("(") + 1, dictObj.indexOf(")") - 1); if (basedir.startsWith( KylinConfig.getInstanceFromEnv().getHdfsWorkingDirectory() + "/resources/GlobalDict")) { activeResources.add(KylinConfig.getInstanceFromEnv().getHdfsWorkingDirectory() + "resources/GlobalDict" + dictInfo.getResourceDir()); } else if (basedir.startsWith(KylinConfig.getInstanceFromEnv().getHdfsWorkingDirectory() + "/resources/SegmentDict")) { activeResources.add(KylinConfig.getInstanceFromEnv().getHdfsWorkingDirectory() + "resources/SegmentDict" + dictInfo.getResourceDir()); } } } } } toDeleteCandidates.keySet().removeAll(activeResources); long outdatedJobTimeCut = System.currentTimeMillis() - jobOutdatedDays * 24 * 3600 * 1000L; ExecutableDao executableDao = ExecutableDao.getInstance(config); List<ExecutablePO> allExecutable = executableDao.getJobs(); for (ExecutablePO executable : allExecutable) { long lastModified = executable.getLastModified(); if (lastModified < outdatedJobTimeCut && isJobComplete(executableDao, executable)) { String jobResPath = ResourceStore.EXECUTE_RESOURCE_ROOT + "/" + executable.getUuid(); String jobOutputResPath = ResourceStore.EXECUTE_OUTPUT_RESOURCE_ROOT + "/" + executable.getUuid(); long outputLastModified = getTimestamp(jobOutputResPath); toDeleteCandidates.put(jobResPath, lastModified); toDeleteCandidates.put(jobOutputResPath, outputLastModified); List<ExecutablePO> tasks = executable.getTasks(); if (tasks != null && !tasks.isEmpty()) { for (ExecutablePO task : executable.getTasks()) { String taskId = task.getUuid(); if (StringUtils.isNotBlank(taskId)) { String resPath = ResourceStore.EXECUTE_OUTPUT_RESOURCE_ROOT + "/" + task.getUuid(); long timestamp = getTimestamp(resPath); toDeleteCandidates.put(resPath, timestamp); } } } } } garbageResources = cleanupConclude(delete, toDeleteCandidates); return garbageResources; } MetadataCleanupJob();  MetadataCleanupJob(KylinConfig config); Map<String, Long> getGarbageResources(); Map<String, Long> cleanup(boolean delete, int jobOutdatedDays);  }
@Test public void testCleanUp() throws Exception { staticCreateTestMetadata(false, new ResetTimeHook(2000, "src/test/resources/test_meta")); MetadataCleanupJob metadataCleanupJob = new MetadataCleanupJob(); Map<String, Long> cleanupMap = metadataCleanupJob.cleanup(false, 30); Assert.assertEquals(7, cleanupMap.size()); for (long timestamp : cleanupMap.values()) { Assert.assertEquals(2000, timestamp); } }
@Test public void testNotCleanUp() throws Exception { staticCreateTestMetadata(false, new ResetTimeHook(System.currentTimeMillis(), "src/test/resources/test_meta")); MetadataCleanupJob metadataCleanupJob = new MetadataCleanupJob(); Map<String, Long> cleanupMap = metadataCleanupJob.cleanup(false, 30); Assert.assertEquals(0, cleanupMap.size()); }