@Test
  public void testWALEntryFilter() throws IOException {
    Configuration conf = HBaseConfiguration.create();
    // Make it so our filter is instantiated on construction of ReplicationSink.
    conf.setClass(HConstants.CLIENT_CONNECTION_REGISTRY_IMPL_CONF_KEY,
      DevNullConnectionRegistry.class, ConnectionRegistry.class);
    conf.setClass(WALEntrySinkFilter.WAL_ENTRY_FILTER_KEY,
      IfTimeIsGreaterThanBOUNDARYWALEntrySinkFilterImpl.class, WALEntrySinkFilter.class);
    conf.setClass(ClusterConnectionFactory.HBASE_SERVER_CLUSTER_CONNECTION_IMPL,
      DevNullAsyncClusterConnection.class, AsyncClusterConnection.class);
    ReplicationSink sink = new ReplicationSink(conf, null);
    // Create some dumb walentries.
    List<AdminProtos.WALEntry> entries = new ArrayList<>();
    AdminProtos.WALEntry.Builder entryBuilder = AdminProtos.WALEntry.newBuilder();
    // Need a tablename.
    ByteString tableName =
      ByteString.copyFromUtf8(TableName.valueOf(this.name.getMethodName()).toString());
    // Add WALEdit Cells to Cells List. The way edits arrive at the sink is with protos
    // describing the edit with all Cells from all edits aggregated in a single CellScanner.
    final List<ExtendedCell> cells = new ArrayList<>();
    int count = BOUNDARY * 2;
    for (int i = 0; i < count; i++) {
      byte[] bytes = Bytes.toBytes(i);
      // Create a wal entry. Everything is set to the current index as bytes or int/long.
      entryBuilder.clear();
      entryBuilder.setKey(entryBuilder.getKeyBuilder().setLogSequenceNumber(i)
        .setEncodedRegionName(ByteString.copyFrom(bytes)).setWriteTime(i).setTableName(tableName)
        .build());
      // Lets have one Cell associated with each WALEdit.
      entryBuilder.setAssociatedCellCount(1);
      entries.add(entryBuilder.build());
      // We need to add a Cell per WALEdit to the cells array.
      ExtendedCellBuilder cellBuilder =
        ExtendedCellBuilderFactory.create(CellBuilderType.DEEP_COPY);
      // Make cells whose row, family, cell, value, and ts are == 'i'.
      ExtendedCell cell = cellBuilder.setRow(bytes).setFamily(bytes).setQualifier(bytes)
        .setType(Cell.Type.Put).setTimestamp(i).setValue(bytes).build();
      cells.add(cell);
    }
    // Now wrap our cells array in a CellScanner that we can pass in to replicateEntries. It has
    // all Cells from all the WALEntries made above.
    ExtendedCellScanner cellScanner = new ExtendedCellScanner() {
      // Set to -1 because advance gets called before current.
      int index = -1;

      @Override
      public ExtendedCell current() {
        return cells.get(index);
      }

      @Override
      public boolean advance() throws IOException {
        index++;
        return index < cells.size();
      }
    };
    // Call our sink.
    sink.replicateEntries(entries, cellScanner, null, null, null);
    // Check what made it through and what was filtered.
    assertTrue(FILTERED.get() > 0);
    assertTrue(UNFILTERED.get() > 0);
    assertEquals(count, FILTERED.get() + UNFILTERED.get());
  }
Test filter. Filter will filter out any write time that is <= 5 (BOUNDARY). We count how many
items we filter out and we count how many cells make it through for distribution way down below
in the Table#batch implementation. Puts in place a custom DevNullConnection so we can insert
our counting Table.
hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestWALEntrySinkFilter.java