@Test
public void testTTLFiltering() throws Exception {

  Map<byte[], Long> ttls = Maps.newTreeMap(Bytes.BYTES_COMPARATOR);
  ttls.put(FAM, 10L);
  ttls.put(FAM2, 30L);
  ttls.put(FAM3, 0L);
  Transaction tx = txManager.startShort();
  long now = tx.getVisibilityUpperBound();
  Filter filter = new TransactionVisibilityFilter(tx, ttls, false, ScanType.USER_SCAN);
  assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL,
               filter.filterKeyValue(newKeyValue("row1", FAM, "val1", now)));
  assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL,
               filter.filterKeyValue(newKeyValue("row1", FAM, "val1", now - 1 * TxConstants.MAX_TX_PER_MS)));
  assertEquals(Filter.ReturnCode.NEXT_COL,
               filter.filterKeyValue(newKeyValue("row1", FAM, "val1", now - 11 * TxConstants.MAX_TX_PER_MS)));
  assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL,
               filter.filterKeyValue(newKeyValue("row1", FAM2, "val1", now - 11 * TxConstants.MAX_TX_PER_MS)));
  assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL,
               filter.filterKeyValue(newKeyValue("row1", FAM2, "val1", now - 21 * TxConstants.MAX_TX_PER_MS)));
  assertEquals(Filter.ReturnCode.NEXT_COL,
               filter.filterKeyValue(newKeyValue("row1", FAM2, "val1", now - 31 * TxConstants.MAX_TX_PER_MS)));
  assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL,
               filter.filterKeyValue(newKeyValue("row1", FAM3, "val1", now - 31 * TxConstants.MAX_TX_PER_MS)));
  assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL,
               filter.filterKeyValue(newKeyValue("row1", FAM3, "val1", now - 1001 * TxConstants.MAX_TX_PER_MS)));
  assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL,
               filter.filterKeyValue(newKeyValue("row2", FAM, "val1", now)));
  assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL,
               filter.filterKeyValue(newKeyValue("row2", FAM, "val1", now - 1 * TxConstants.MAX_TX_PER_MS)));
  // Verify ttl for pre-existing, non-transactional data
  long preNow = now / TxConstants.MAX_TX_PER_MS;
  assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL,
               filter.filterKeyValue(newKeyValue("row1", FAM, "val1", preNow)));
  assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL,
               filter.filterKeyValue(newKeyValue("row1", FAM, "val1", preNow - 9L)));
  assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL,
               filter.filterKeyValue(newKeyValue("row1", FAM, "val1", preNow - 10L)));
  assertEquals(Filter.ReturnCode.NEXT_COL,
               filter.filterKeyValue(newKeyValue("row1", FAM, "val1", preNow - 11L)));
  assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL,
               filter.filterKeyValue(newKeyValue("row1", FAM3, "val1", preNow)));
  assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL,
               filter.filterKeyValue(newKeyValue("row1", FAM3, "val1", preNow - 9L)));
  assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL,
               filter.filterKeyValue(newKeyValue("row1", FAM3, "val1", preNow - 10L)));
  assertEquals(Filter.ReturnCode.INCLUDE_AND_NEXT_COL,
               filter.filterKeyValue(newKeyValue("row1", FAM3, "val1", preNow - 1001L)));
}