WAL implements Closeable { private void roll() throws IOException { try { rollInProgress = true; LOG.info("Rolling WAL " + this.path); if (dataFileWALWriter != null) { fileLargestSequenceIDMap.put(dataFileWALWriter.getPath() .getAbsolutePath(), dataFileWALWriter.getLargestSequenceID()); dataFileWALWriter.close(); } if (sequenceIDWALWriter != null) { fileLargestSequenceIDMap.put(sequenceIDWALWriter.getPath() .getAbsolutePath(), sequenceIDWALWriter.getLargestSequenceID()); sequenceIDWALWriter.close(); } long ts = System.currentTimeMillis(); File dataWalFileName = new File(dataPath, Long.toString(ts)); File seqWalFileName = new File(sequenceIDPath, Long.toString(ts)); while (dataWalFileName.exists() || seqWalFileName.exists()) { ts++; dataWalFileName = new File(dataPath, Long.toString(ts)); seqWalFileName = new File(sequenceIDPath, Long.toString(ts)); } dataFileWALWriter = new WALDataFile.Writer<T>(dataWalFileName); sequenceIDWALWriter = new WALDataFile.Writer<NullWritable>(seqWalFileName); rollRequired = false; } finally { rollInProgress = false; synchronized (this) { notifyAll(); } } } WAL(File path, Class<T> clazz);  WAL(File path, Class<T> clazz, long rollSize,
      long maxLogsSize, long minLogRentionPeriod,
      long workerInterval); WALReplayResult<T> replay(); void writeEntries(List<WALEntry<T>> entries); void writeEntry(WALEntry<T> entry); void writeSequenceID(long sequenceID); void writeSequenceIDs(List<Long> sequenceIDs); @Override void close(); void setRollSize(long rollSize); void setMaxLogsSize(long maxLogsSize); void setMinLogRetentionPeriod(long minLogRetentionPeriod); void setWorkerInterval(long workerInterval); @SuppressWarnings({ "unchecked", "rawtypes" }) static void main(String[] args); static final long DEFAULT_ROLL_SIZE; static final long DEFAULT_MAX_LOGS_SIZE; static final long DEFAULT_MIN_LOG_RENTENTION_PERIOD; static final long DEFAULT_WORKER_INTERVAL; }
@Test public void testRoll() throws IOException, InterruptedException { wal.close(); wal = new WAL<Text>(dataDir, Text.class, 0L, 0L, 0L, 1L); long seqid = 0; List<String> expected = strings(100); for(String s : expected) { wal.writeEntry(new WALEntry<Text>(new Text(s), ++seqid)); Thread.sleep(1); wal.writeSequenceID(seqid); Thread.sleep(1); } wal.writeSequenceID(Long.MAX_VALUE); Thread.sleep(1000L); wal.close(); File seq = new File(dataDir, "seq"); File[] seqFiles = seq.listFiles(); Assert.assertNotNull(seqFiles); Assert.assertTrue(seqFiles.length < 5); File data = new File(dataDir, "data"); File[] dataFiles = data.listFiles(); Assert.assertNotNull(dataFiles); Assert.assertTrue(dataFiles.length < 5); }