class 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;
  }

  public  CheckpointRebuilder(List<File> logFiles, FlumeEventQueue queue,
                             boolean fsyncPerTransaction);

}