@Test public void recover() throws IOException, StorageEngineException, InterruptedException, IllegalPathException { receiverLogger = new SyncReceiverLogger( new File(getReceiverFolderFile(), SyncConstant.SYNC_LOG_NAME)); fileLoader = FileLoader.createFileLoader(getReceiverFolderFile()); Map<String, Set<File>> allFileList = new HashMap<>(); Map<String, Set<File>> correctSequenceLoadedFileMap = new HashMap<>(); Random r = new Random(0); receiverLogger.startSyncTsFiles(); Set<String> toBeSyncedFiles = new HashSet<>(); for (int i = 0; i < 3; i++) { for (int j = 0; j < 10; j++) { allFileList.putIfAbsent(SG_NAME + i, new HashSet<>()); correctSequenceLoadedFileMap.putIfAbsent(SG_NAME + i, new HashSet<>()); String rand = String.valueOf(r.nextInt(10000) + i * j); String fileName = getSnapshotFolder() + File.separator + SG_NAME + i + File.separator + System .currentTimeMillis() + IoTDBConstant.FILE_NAME_SEPARATOR + rand + IoTDBConstant.FILE_NAME_SEPARATOR + "0.tsfile"; Thread.sleep(1); File syncFile = new File(fileName); receiverLogger .finishSyncTsfile(syncFile); toBeSyncedFiles.add(syncFile.getAbsolutePath()); File dataFile = new File( DirectoryManager.getInstance().getNextFolderForSequenceFile(), syncFile.getParentFile().getName() + File.separatorChar + syncFile.getName()); correctSequenceLoadedFileMap.get(SG_NAME + i).add(dataFile); allFileList.get(SG_NAME + i).add(syncFile); if (!syncFile.getParentFile().exists()) { syncFile.getParentFile().mkdirs(); } if (!syncFile.exists() && !syncFile.createNewFile()) { LOGGER.error("Can not create new file {}", syncFile.getPath()); } if (!new File(syncFile.getAbsolutePath() + TsFileResource.RESOURCE_SUFFIX).exists() && !new File(syncFile.getAbsolutePath() + TsFileResource.RESOURCE_SUFFIX) .createNewFile()) { LOGGER.error("Can not create new file {}", syncFile.getPath()); } TsFileResource tsFileResource = new TsFileResource(syncFile); tsFileResource.putStartTime(String.valueOf(i), (long) j * 10); tsFileResource.putEndTime(String.valueOf(i), (long) j * 10 + 5); tsFileResource.serialize(); } } for (int i = 0; i < 3; i++) { StorageGroupProcessor processor = StorageEngine.getInstance().getProcessor(new PartialPath(SG_NAME + i)); assertTrue(processor.getSequenceFileTreeSet().isEmpty()); assertTrue(processor.getUnSequenceFileList().isEmpty()); } assertTrue(getReceiverFolderFile().exists()); for (Set<File> set : allFileList.values()) { for (File newTsFile : set) { if (!newTsFile.getName().endsWith(TsFileResource.RESOURCE_SUFFIX)) { fileLoader.addTsfile(newTsFile); } } } receiverLogger.close(); assertTrue(new File(getReceiverFolderFile(), SyncConstant.LOAD_LOG_NAME).exists()); assertTrue(new File(getReceiverFolderFile(), SyncConstant.SYNC_LOG_NAME).exists()); assertTrue( FileLoaderManager.getInstance().containsFileLoader(getReceiverFolderFile().getName())); int mode = 0; Set<String> toBeSyncedFilesTest = new HashSet<>(); try (BufferedReader br = new BufferedReader( new FileReader(new File(getReceiverFolderFile(), SyncConstant.SYNC_LOG_NAME)))) { String line; while ((line = br.readLine()) != null) { if (line.equals(SyncReceiverLogger.SYNC_DELETED_FILE_NAME_START)) { mode = -1; } else if (line.equals(SyncReceiverLogger.SYNC_TSFILE_START)) { mode = 1; } else { if (mode == 1) { toBeSyncedFilesTest.add(line); } } } } assertEquals(toBeSyncedFilesTest.size(), toBeSyncedFiles.size()); assertTrue(toBeSyncedFilesTest.containsAll(toBeSyncedFiles)); logAnalyze.recover(getReceiverFolderFile().getName()); try { long waitTime = 0; while (FileLoaderManager.getInstance() .containsFileLoader(getReceiverFolderFile().getName())) { Thread.sleep(100); waitTime += 100; LOGGER.info("Has waited for loading new tsfiles {}ms", waitTime); } } catch (InterruptedException e) { LOGGER.error("Fail to wait for loading new tsfiles", e); Thread.currentThread().interrupt(); throw e; } assertFalse(new File(getReceiverFolderFile(), SyncConstant.LOAD_LOG_NAME).exists()); assertFalse(new File(getReceiverFolderFile(), SyncConstant.SYNC_LOG_NAME).exists()); }
private boolean recover(File senderFolder) throws IOException { if (!new File(senderFolder, SyncConstant.SYNC_LOG_NAME).exists()) { new File(senderFolder, SyncConstant.LOAD_LOG_NAME).delete(); FileUtils.deleteDirectory(new File(senderFolder, SyncConstant.RECEIVER_DATA_FOLDER_NAME)); return true; } if (FileLoaderManager.getInstance().containsFileLoader(senderFolder.getName())) { FileLoaderManager.getInstance().getFileLoader(senderFolder.getName()).endSync(); try { Thread.sleep(FileLoader.WAIT_TIME << 1); } catch (InterruptedException e) { LOGGER.error("Thread is interrupted from waiting for ending sync in recovery."); Thread.currentThread().interrupt(); } } else { scanLogger(FileLoader.createFileLoader(senderFolder), new File(senderFolder, SyncConstant.SYNC_LOG_NAME), new File(senderFolder, SyncConstant.LOAD_LOG_NAME)); } return !FileLoaderManager.getInstance().containsFileLoader(senderFolder.getName()); }
SyncReceiverLogAnalyzer implements ISyncReceiverLogAnalyzer { private boolean recover(File senderFolder) throws IOException { if (!new File(senderFolder, SyncConstant.SYNC_LOG_NAME).exists()) { new File(senderFolder, SyncConstant.LOAD_LOG_NAME).delete(); FileUtils.deleteDirectory(new File(senderFolder, SyncConstant.RECEIVER_DATA_FOLDER_NAME)); return true; } if (FileLoaderManager.getInstance().containsFileLoader(senderFolder.getName())) { FileLoaderManager.getInstance().getFileLoader(senderFolder.getName()).endSync(); try { Thread.sleep(FileLoader.WAIT_TIME << 1); } catch (InterruptedException e) { LOGGER.error("Thread is interrupted from waiting for ending sync in recovery."); Thread.currentThread().interrupt(); } } else { scanLogger(FileLoader.createFileLoader(senderFolder), new File(senderFolder, SyncConstant.SYNC_LOG_NAME), new File(senderFolder, SyncConstant.LOAD_LOG_NAME)); } return !FileLoaderManager.getInstance().containsFileLoader(senderFolder.getName()); } }
SyncReceiverLogAnalyzer implements ISyncReceiverLogAnalyzer { private boolean recover(File senderFolder) throws IOException { if (!new File(senderFolder, SyncConstant.SYNC_LOG_NAME).exists()) { new File(senderFolder, SyncConstant.LOAD_LOG_NAME).delete(); FileUtils.deleteDirectory(new File(senderFolder, SyncConstant.RECEIVER_DATA_FOLDER_NAME)); return true; } if (FileLoaderManager.getInstance().containsFileLoader(senderFolder.getName())) { FileLoaderManager.getInstance().getFileLoader(senderFolder.getName()).endSync(); try { Thread.sleep(FileLoader.WAIT_TIME << 1); } catch (InterruptedException e) { LOGGER.error("Thread is interrupted from waiting for ending sync in recovery."); Thread.currentThread().interrupt(); } } else { scanLogger(FileLoader.createFileLoader(senderFolder), new File(senderFolder, SyncConstant.SYNC_LOG_NAME), new File(senderFolder, SyncConstant.LOAD_LOG_NAME)); } return !FileLoaderManager.getInstance().containsFileLoader(senderFolder.getName()); } private  SyncReceiverLogAnalyzer(); }
SyncReceiverLogAnalyzer implements ISyncReceiverLogAnalyzer { private boolean recover(File senderFolder) throws IOException { if (!new File(senderFolder, SyncConstant.SYNC_LOG_NAME).exists()) { new File(senderFolder, SyncConstant.LOAD_LOG_NAME).delete(); FileUtils.deleteDirectory(new File(senderFolder, SyncConstant.RECEIVER_DATA_FOLDER_NAME)); return true; } if (FileLoaderManager.getInstance().containsFileLoader(senderFolder.getName())) { FileLoaderManager.getInstance().getFileLoader(senderFolder.getName()).endSync(); try { Thread.sleep(FileLoader.WAIT_TIME << 1); } catch (InterruptedException e) { LOGGER.error("Thread is interrupted from waiting for ending sync in recovery."); Thread.currentThread().interrupt(); } } else { scanLogger(FileLoader.createFileLoader(senderFolder), new File(senderFolder, SyncConstant.SYNC_LOG_NAME), new File(senderFolder, SyncConstant.LOAD_LOG_NAME)); } return !FileLoaderManager.getInstance().containsFileLoader(senderFolder.getName()); } private  SyncReceiverLogAnalyzer(); static SyncReceiverLogAnalyzer getInstance(); @Override void recoverAll(); @Override boolean recover(String senderName); @Override void scanLogger(IFileLoader loader, File syncLog, File loadLog); }
SyncReceiverLogAnalyzer implements ISyncReceiverLogAnalyzer { private boolean recover(File senderFolder) throws IOException { if (!new File(senderFolder, SyncConstant.SYNC_LOG_NAME).exists()) { new File(senderFolder, SyncConstant.LOAD_LOG_NAME).delete(); FileUtils.deleteDirectory(new File(senderFolder, SyncConstant.RECEIVER_DATA_FOLDER_NAME)); return true; } if (FileLoaderManager.getInstance().containsFileLoader(senderFolder.getName())) { FileLoaderManager.getInstance().getFileLoader(senderFolder.getName()).endSync(); try { Thread.sleep(FileLoader.WAIT_TIME << 1); } catch (InterruptedException e) { LOGGER.error("Thread is interrupted from waiting for ending sync in recovery."); Thread.currentThread().interrupt(); } } else { scanLogger(FileLoader.createFileLoader(senderFolder), new File(senderFolder, SyncConstant.SYNC_LOG_NAME), new File(senderFolder, SyncConstant.LOAD_LOG_NAME)); } return !FileLoaderManager.getInstance().containsFileLoader(senderFolder.getName()); } private  SyncReceiverLogAnalyzer(); static SyncReceiverLogAnalyzer getInstance(); @Override void recoverAll(); @Override boolean recover(String senderName); @Override void scanLogger(IFileLoader loader, File syncLog, File loadLog);  }