@Test
  public void testPartialRead() throws Exception {
    final WALFactory walfactory = new WALFactory(conf, getName());
    WAL log = walfactory.getWAL(info);
    // This test depends on timestamp being millisecond based and the filename of the WAL also
    // being millisecond based.
    long ts = EnvironmentEdgeManager.currentTime();
    WALEdit edit = new WALEdit();
    WALEditInternalHelper.addExtendedCell(edit,
      new KeyValue(rowName, family, Bytes.toBytes("1"), ts, value));
    log.appendData(info, getWalKeyImpl(ts, scopes), edit);
    edit = new WALEdit();
    WALEditInternalHelper.addExtendedCell(edit,
      new KeyValue(rowName, family, Bytes.toBytes("2"), ts + 1, value));
    log.appendData(info, getWalKeyImpl(ts + 1, scopes), edit);
    log.sync();
    Threads.sleep(10);
    LOG.info("Before 1st WAL roll " + log.toString());
    log.rollWriter();
    LOG.info("Past 1st WAL roll " + log.toString());

    Thread.sleep(1);
    long ts1 = EnvironmentEdgeManager.currentTime();

    edit = new WALEdit();
    WALEditInternalHelper.addExtendedCell(edit,
      new KeyValue(rowName, family, Bytes.toBytes("3"), ts1 + 1, value));
    log.appendData(info, getWalKeyImpl(ts1 + 1, scopes), edit);
    edit = new WALEdit();
    WALEditInternalHelper.addExtendedCell(edit,
      new KeyValue(rowName, family, Bytes.toBytes("4"), ts1 + 2, value));
    log.appendData(info, getWalKeyImpl(ts1 + 2, scopes), edit);
    log.sync();
    log.shutdown();
    walfactory.shutdown();
    LOG.info("Closed WAL " + log.toString());

    WALInputFormat input = new WALInputFormat();
    Configuration jobConf = new Configuration(conf);
    jobConf.set("mapreduce.input.fileinputformat.inputdir", logDir.toString());
    jobConf.setLong(WALInputFormat.END_TIME_KEY, ts);

    // Only 1st file is considered, and only its 1st entry is in-range.
    List<InputSplit> splits = input.getSplits(MapreduceTestingShim.createJobContext(jobConf));
    assertEquals(1, splits.size());
    testSplit(splits.get(0), Bytes.toBytes("1"));

    jobConf.setLong(WALInputFormat.END_TIME_KEY, ts1 + 1);
    splits = input.getSplits(MapreduceTestingShim.createJobContext(jobConf));
    assertEquals(2, splits.size());
    // Both entries from first file are in-range.
    testSplit(splits.get(0), Bytes.toBytes("1"), Bytes.toBytes("2"));
    // Only the 1st entry from the 2nd file is in-range.
    testSplit(splits.get(1), Bytes.toBytes("3"));

    jobConf.setLong(WALInputFormat.START_TIME_KEY, ts + 1);
    jobConf.setLong(WALInputFormat.END_TIME_KEY, ts1 + 1);
    splits = input.getSplits(MapreduceTestingShim.createJobContext(jobConf));
    assertEquals(1, splits.size());
    // Only the 1st entry from the 2nd file is in-range.
    testSplit(splits.get(0), Bytes.toBytes("3"));
  }
Test partial reads from the WALs based on passed time range.
hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestWALRecordReader.java