CheckpointRebuilder { public boolean rebuild() throws IOException, Exception { LOG.info("Attempting to fast replay the log files."); List<LogFile.SequentialReader> logReaders = Lists.newArrayList(); for (File logFile : logFiles) { try { logReaders.add(LogFileFactory.getSequentialReader(logFile, null, fsyncPerTransaction)); } catch (EOFException e) { LOG.warn("Ignoring " + logFile + " due to EOF", e); } } long transactionIDSeed = 0; long writeOrderIDSeed = 0; try { for (LogFile.SequentialReader log : logReaders) { LogRecord entry; int fileID = log.getLogFileID(); while ((entry = log.next()) != null) { int offset = entry.getOffset(); TransactionEventRecord record = entry.getEvent(); long trans = record.getTransactionID(); long writeOrderID = record.getLogWriteOrderID(); transactionIDSeed = Math.max(trans, transactionIDSeed); writeOrderIDSeed = Math.max(writeOrderID, writeOrderIDSeed); if (record.getRecordType() == TransactionEventRecord.Type.PUT.get()) { uncommittedPuts.put(record.getTransactionID(), new ComparableFlumeEventPointer( new FlumeEventPointer(fileID, offset), record.getLogWriteOrderID())); } else if (record.getRecordType() == TransactionEventRecord.Type.TAKE.get()) { Take take = (Take) record; uncommittedTakes.put(record.getTransactionID(), new ComparableFlumeEventPointer( new FlumeEventPointer(take.getFileID(), take.getOffset()), record.getLogWriteOrderID())); } else if (record.getRecordType() == TransactionEventRecord.Type.COMMIT.get()) { Commit commit = (Commit) record; if (commit.getType() == TransactionEventRecord.Type.PUT.get()) { Set<ComparableFlumeEventPointer> puts = uncommittedPuts.get(record.getTransactionID()); if (puts != null) { for (ComparableFlumeEventPointer put : puts) { if (!pendingTakes.remove(put)) { committedPuts.add(put); } } } } else { Set<ComparableFlumeEventPointer> takes = uncommittedTakes.get(record.getTransactionID()); if (takes != null) { for (ComparableFlumeEventPointer take : takes) { if (!committedPuts.remove(take)) { pendingTakes.add(take); } } } } } else if (record.getRecordType() == TransactionEventRecord.Type.ROLLBACK.get()) { if (uncommittedPuts.containsKey(record.getTransactionID())) { uncommittedPuts.removeAll(record.getTransactionID()); } else { uncommittedTakes.removeAll(record.getTransactionID()); } } } } } catch (Exception e) { LOG.warn("Error while generating checkpoint using fast generation logic", e); return false; } finally { TransactionIDOracle.setSeed(transactionIDSeed); WriteOrderOracle.setSeed(writeOrderIDSeed); for (LogFile.SequentialReader reader : logReaders) { reader.close(); } } Set<ComparableFlumeEventPointer> sortedPuts = Sets.newTreeSet(committedPuts); int count = 0; for (ComparableFlumeEventPointer put : sortedPuts) { queue.addTail(put.pointer); count++; } LOG.info("Replayed {} events using fast replay logic.", count); return true; } CheckpointRebuilder(List<File> logFiles, FlumeEventQueue queue,
                             boolean fsyncPerTransaction); boolean rebuild(); static void main(String[] args);  }
@Test public void testFastReplay() throws Exception { Map<String, String> overrides = Maps.newHashMap(); overrides.put(FileChannelConfiguration.CAPACITY, String.valueOf(50)); overrides.put(FileChannelConfiguration.TRANSACTION_CAPACITY, String.valueOf(50)); channel = createFileChannel(overrides); channel.start(); Assert.assertTrue(channel.isOpen()); Set<String> in = fillChannel(channel, "checkpointBulder"); channel.stop(); File checkpointFile = new File(checkpointDir, "checkpoint"); File metaDataFile = Serialization.getMetaDataFile(checkpointFile); File inflightTakesFile = new File(checkpointDir, "inflighttakes"); File inflightPutsFile = new File(checkpointDir, "inflightputs"); File queueSetDir = new File(checkpointDir, "queueset"); Assert.assertTrue(checkpointFile.delete()); Assert.assertTrue(metaDataFile.delete()); Assert.assertTrue(inflightTakesFile.delete()); Assert.assertTrue(inflightPutsFile.delete()); EventQueueBackingStore backingStore = EventQueueBackingStoreFactory.get(checkpointFile, 50, "test", new FileChannelCounter("test")); FlumeEventQueue queue = new FlumeEventQueue(backingStore, inflightTakesFile, inflightPutsFile, queueSetDir); CheckpointRebuilder checkpointRebuilder = new CheckpointRebuilder(getAllLogs(dataDirs), queue, true); Assert.assertTrue(checkpointRebuilder.rebuild()); channel = createFileChannel(overrides); channel.start(); Assert.assertTrue(channel.isOpen()); Set<String> out = consumeChannel(channel); compareInputAndOut(in, out); }