@Test
public void testEventFilter() throws Exception {

  TableCache tableCache = mock(TableCache.class);
  BinlogEvent lastEvent = new XidEvent(0l, 0l, BINLOG_FILE_POS, 0l);
  BinlogFilePos nextPosition = new BinlogFilePos("test.123", 15, 100);
  MysqlSourceState state = new MysqlSourceState(0l, lastEvent.getOffset(), 0l, BINLOG_FILE_POS);
  Filter<BinlogEvent> filter =
      MysqlEventFilter.create(tableCache, TABLE_NAMES, new AtomicReference(state));
  when(tableCache.contains(TABLE_ID)).thenReturn(true);
  assertTrue(
      filter.apply(
          new TableMapEvent(
              TABLE_ID, 0l, 0l, nextPosition, DATABASE_NAME, TABLE_NAME, new byte[1])));
  assertTrue(
      filter.apply(new WriteEvent(TABLE_ID, 0l, 0l, nextPosition, Collections.emptyList())));
  assertTrue(
      filter.apply(new DeleteEvent(TABLE_ID, 0l, 0l, nextPosition, Collections.emptyList())));
  assertTrue(
      filter.apply(new UpdateEvent(TABLE_ID, 0l, 0l, nextPosition, Collections.emptyList())));
  assertTrue(filter.apply(new XidEvent(0l, 0l, BINLOG_FILE_POS, 12l)));
  assertTrue(filter.apply(new QueryEvent(0l, 0l, BINLOG_FILE_POS, DATABASE_NAME, "")));
  assertTrue(filter.apply(new StartEvent(0l, 0l, BINLOG_FILE_POS)));
  assertFalse(
      filter.apply(new TableMapEvent(TABLE_ID, 0l, 0l, BINLOG_FILE_POS, "", "", new byte[1])));
  assertFalse(filter.apply(new WriteEvent(0l, 0l, 0l, BINLOG_FILE_POS, Collections.emptyList())));
  assertFalse(
      filter.apply(new WriteEvent(TABLE_ID, 0l, 0l, BINLOG_FILE_POS, Collections.emptyList())));
  assertFalse(filter.apply(mock(BinlogEvent.class)));
}