ThemisExpiredDataCleanFilter extends FilterBase { public ThemisExpiredDataCleanFilter() { this.region = null; } ThemisExpiredDataCleanFilter();  ThemisExpiredDataCleanFilter(long cleanTs);  ThemisExpiredDataCleanFilter(long cleanTs, HRegion region); @Override ReturnCode filterKeyValue(KeyValue kv); void readFields(DataInput in); void write(DataOutput out);  }
@Test public void testThemisExpiredDataCleanFilter() throws IOException { writeData(COLUMN, prewriteTs); writeData(COLUMN, prewriteTs + 1); ResultScanner scanner = getResultScanner(TABLENAME, prewriteTs); Result result = scanner.next(); Assert.assertEquals(2, result.size()); Assert.assertEquals(prewriteTs + 1, result.list().get(0).getTimestamp()); Assert.assertEquals(prewriteTs, result.list().get(1).getTimestamp()); scanner.close(); scanner = getResultScanner(TABLENAME, prewriteTs + 1); result = scanner.next(); Assert.assertEquals(2, result.size()); Assert.assertEquals(prewriteTs + 1, result.list().get(0).getTimestamp()); Assert.assertEquals(prewriteTs, result.list().get(1).getTimestamp()); scanner.close(); scanner = getResultScanner(TABLENAME, prewriteTs + 2); result = scanner.next(); Assert.assertEquals(1, result.size()); Assert.assertEquals(prewriteTs + 1, result.list().get(0).getTimestamp()); scanner.close(); writeData(COLUMN_WITH_ANOTHER_QUALIFIER, prewriteTs); writeData(COLUMN_WITH_ANOTHER_QUALIFIER, prewriteTs + 1); writeData(COLUMN_WITH_ANOTHER_FAMILY, prewriteTs); writeData(COLUMN_WITH_ANOTHER_FAMILY, prewriteTs + 1); scanner = getResultScanner(TABLENAME, prewriteTs + 2); result = scanner.next(); scanner.close(); Assert.assertEquals(3, result.size()); ColumnCoordinate[] columns = new ColumnCoordinate[] { COLUMN_WITH_ANOTHER_FAMILY, COLUMN_WITH_ANOTHER_QUALIFIER, COLUMN }; for (int i = 0; i < columns.length; ++i) { ColumnCoordinate column = columns[i]; KeyValue kv = result.list().get(i); Assert.assertEquals(prewriteTs + 1, kv.getTimestamp()); Assert.assertArrayEquals(column.getFamily(), kv.getFamily()); Assert.assertArrayEquals(column.getQualifier(), kv.getQualifier()); } writeData(COLUMN_WITH_ANOTHER_ROW, prewriteTs); writeData(COLUMN_WITH_ANOTHER_ROW, prewriteTs + 1); scanner = getResultScanner(TABLENAME, prewriteTs + 2); result = scanner.next(); Assert.assertEquals(1, result.size()); Assert.assertEquals(prewriteTs + 1, result.list().get(0).getTimestamp()); result = scanner.next(); Assert.assertEquals(3, result.size()); for (int i = 0; i < columns.length; ++i) { ColumnCoordinate column = columns[i]; KeyValue kv = result.list().get(i); Assert.assertEquals(prewriteTs + 1, kv.getTimestamp()); Assert.assertArrayEquals(column.getFamily(), kv.getFamily()); Assert.assertArrayEquals(column.getQualifier(), kv.getQualifier()); } scanner.close(); }