@Test public void testRecovery() throws OverflowProcessorException, IOException { processor = new OverflowProcessor(processorName, parameters, OverflowTestUtils.getFileSchema()); processor.close(); processor.switchWorkToMerge(); assertEquals(true, processor.isMerge()); processor.clear(); OverflowProcessor overflowProcessor = new OverflowProcessor(processorName, parameters, OverflowTestUtils.getFileSchema()); assertEquals(false, overflowProcessor.isMerge()); overflowProcessor.switchWorkToMerge(); OverflowSeriesDataSource overflowSeriesDataSource = overflowProcessor .query(OverflowTestUtils.deviceId1, OverflowTestUtils.measurementId1, null, OverflowTestUtils.dataType1); Assert.assertEquals(true, overflowSeriesDataSource.getReadableMemChunk().isEmpty()); assertEquals(2, overflowSeriesDataSource.getOverflowInsertFileList().size()); overflowProcessor.switchMergeToWork(); overflowProcessor.close(); overflowProcessor.clear(); }
private void recovery(File parentFile) throws IOException { String[] subFilePaths = clearFile(parentFile.list()); if (subFilePaths.length == 0) { workResource = new OverflowResource(parentPath, String.valueOf(dataPahtCount.getAndIncrement())); return; } else if (subFilePaths.length == 1) { long count = Long.valueOf(subFilePaths[0]); dataPahtCount.addAndGet(count + 1); workResource = new OverflowResource(parentPath, String.valueOf(count)); LOGGER.info("The overflow processor {} recover from work status.", getProcessorName()); } else { long count1 = Long.valueOf(subFilePaths[0]); long count2 = Long.valueOf(subFilePaths[1]); if (count1 > count2) { long temp = count1; count1 = count2; count2 = temp; } dataPahtCount.addAndGet(count2 + 1); workResource = new OverflowResource(parentPath, String.valueOf(count2)); mergeResource = new OverflowResource(parentPath, String.valueOf(count1)); LOGGER.info("The overflow processor {} recover from merge status.", getProcessorName()); } }
OverflowProcessor extends Processor { private void recovery(File parentFile) throws IOException { String[] subFilePaths = clearFile(parentFile.list()); if (subFilePaths.length == 0) { workResource = new OverflowResource(parentPath, String.valueOf(dataPahtCount.getAndIncrement())); return; } else if (subFilePaths.length == 1) { long count = Long.valueOf(subFilePaths[0]); dataPahtCount.addAndGet(count + 1); workResource = new OverflowResource(parentPath, String.valueOf(count)); LOGGER.info("The overflow processor {} recover from work status.", getProcessorName()); } else { long count1 = Long.valueOf(subFilePaths[0]); long count2 = Long.valueOf(subFilePaths[1]); if (count1 > count2) { long temp = count1; count1 = count2; count2 = temp; } dataPahtCount.addAndGet(count2 + 1); workResource = new OverflowResource(parentPath, String.valueOf(count2)); mergeResource = new OverflowResource(parentPath, String.valueOf(count1)); LOGGER.info("The overflow processor {} recover from merge status.", getProcessorName()); } } }
OverflowProcessor extends Processor { private void recovery(File parentFile) throws IOException { String[] subFilePaths = clearFile(parentFile.list()); if (subFilePaths.length == 0) { workResource = new OverflowResource(parentPath, String.valueOf(dataPahtCount.getAndIncrement())); return; } else if (subFilePaths.length == 1) { long count = Long.valueOf(subFilePaths[0]); dataPahtCount.addAndGet(count + 1); workResource = new OverflowResource(parentPath, String.valueOf(count)); LOGGER.info("The overflow processor {} recover from work status.", getProcessorName()); } else { long count1 = Long.valueOf(subFilePaths[0]); long count2 = Long.valueOf(subFilePaths[1]); if (count1 > count2) { long temp = count1; count1 = count2; count2 = temp; } dataPahtCount.addAndGet(count2 + 1); workResource = new OverflowResource(parentPath, String.valueOf(count2)); mergeResource = new OverflowResource(parentPath, String.valueOf(count1)); LOGGER.info("The overflow processor {} recover from merge status.", getProcessorName()); } } OverflowProcessor(String processorName, Map<String, Action> parameters,
      FileSchema fileSchema); }
OverflowProcessor extends Processor { private void recovery(File parentFile) throws IOException { String[] subFilePaths = clearFile(parentFile.list()); if (subFilePaths.length == 0) { workResource = new OverflowResource(parentPath, String.valueOf(dataPahtCount.getAndIncrement())); return; } else if (subFilePaths.length == 1) { long count = Long.valueOf(subFilePaths[0]); dataPahtCount.addAndGet(count + 1); workResource = new OverflowResource(parentPath, String.valueOf(count)); LOGGER.info("The overflow processor {} recover from work status.", getProcessorName()); } else { long count1 = Long.valueOf(subFilePaths[0]); long count2 = Long.valueOf(subFilePaths[1]); if (count1 > count2) { long temp = count1; count1 = count2; count2 = temp; } dataPahtCount.addAndGet(count2 + 1); workResource = new OverflowResource(parentPath, String.valueOf(count2)); mergeResource = new OverflowResource(parentPath, String.valueOf(count1)); LOGGER.info("The overflow processor {} recover from merge status.", getProcessorName()); } } OverflowProcessor(String processorName, Map<String, Action> parameters,
      FileSchema fileSchema); void insert(TSRecord tsRecord); @Deprecated void update(String deviceId, String measurementId, long startTime, long endTime,
      TSDataType type,
      byte[] value); @Deprecated void update(String deviceId, String measurementId, long startTime, long endTime,
      TSDataType type,
      String value); @Deprecated void delete(String deviceId, String measurementId, long timestamp, TSDataType type); OverflowSeriesDataSource query(String deviceId, String measurementId, Filter filter,
      TSDataType dataType); MergeSeriesDataSource queryMerge(String deviceId, String measurementId,
      TSDataType dataType); OverflowSeriesDataSource queryMerge(String deviceId, String measurementId,
      TSDataType dataType,
      boolean isMerge); void switchWorkToMerge(); void switchMergeToWork(); boolean isMerge(); boolean isFlush(); @Override boolean flush(); @Override void close(); void clear(); @Override boolean canBeClosed(); @Override long memoryUsage(); String getOverflowRestoreFile(); long getMetaSize(); long getFileSize(); WriteLogNode getLogNode(); OverflowResource getWorkResource(); }
OverflowProcessor extends Processor { private void recovery(File parentFile) throws IOException { String[] subFilePaths = clearFile(parentFile.list()); if (subFilePaths.length == 0) { workResource = new OverflowResource(parentPath, String.valueOf(dataPahtCount.getAndIncrement())); return; } else if (subFilePaths.length == 1) { long count = Long.valueOf(subFilePaths[0]); dataPahtCount.addAndGet(count + 1); workResource = new OverflowResource(parentPath, String.valueOf(count)); LOGGER.info("The overflow processor {} recover from work status.", getProcessorName()); } else { long count1 = Long.valueOf(subFilePaths[0]); long count2 = Long.valueOf(subFilePaths[1]); if (count1 > count2) { long temp = count1; count1 = count2; count2 = temp; } dataPahtCount.addAndGet(count2 + 1); workResource = new OverflowResource(parentPath, String.valueOf(count2)); mergeResource = new OverflowResource(parentPath, String.valueOf(count1)); LOGGER.info("The overflow processor {} recover from merge status.", getProcessorName()); } } OverflowProcessor(String processorName, Map<String, Action> parameters,
      FileSchema fileSchema); void insert(TSRecord tsRecord); @Deprecated void update(String deviceId, String measurementId, long startTime, long endTime,
      TSDataType type,
      byte[] value); @Deprecated void update(String deviceId, String measurementId, long startTime, long endTime,
      TSDataType type,
      String value); @Deprecated void delete(String deviceId, String measurementId, long timestamp, TSDataType type); OverflowSeriesDataSource query(String deviceId, String measurementId, Filter filter,
      TSDataType dataType); MergeSeriesDataSource queryMerge(String deviceId, String measurementId,
      TSDataType dataType); OverflowSeriesDataSource queryMerge(String deviceId, String measurementId,
      TSDataType dataType,
      boolean isMerge); void switchWorkToMerge(); void switchMergeToWork(); boolean isMerge(); boolean isFlush(); @Override boolean flush(); @Override void close(); void clear(); @Override boolean canBeClosed(); @Override long memoryUsage(); String getOverflowRestoreFile(); long getMetaSize(); long getFileSize(); WriteLogNode getLogNode(); OverflowResource getWorkResource();  }