ThemisRegionObserver extends BaseRegionObserver { @Override public InternalScanner preFlushScannerOpen(final ObserverContext<RegionCoprocessorEnvironment> c, final Store store, final KeyValueScanner memstoreScanner, final InternalScanner s) throws IOException { if (expiredDataCleanEnable && (ThemisMasterObserver.isThemisEnableFamily(store.getFamily()) || ColumnUtil .isCommitFamily(store.getFamily().getName()))) { InternalScanner scanner = getScannerToCleanExpiredThemisData(store, store.scanInfo, Collections.singletonList(memstoreScanner), ScanType.MINOR_COMPACT, store.getHRegion() .getSmallestReadPoint(), HConstants.OLDEST_TIMESTAMP, false); if (scanner != null) { return scanner; } } return s; }  @Override void start(CoprocessorEnvironment e); @Override void prePut(final ObserverContext<RegionCoprocessorEnvironment> c, final Put put,
      final WALEdit edit, final boolean writeToWAL); @Override InternalScanner preFlushScannerOpen(final ObserverContext<RegionCoprocessorEnvironment> c,
      final Store store, final KeyValueScanner memstoreScanner, final InternalScanner s); @Override InternalScanner preCompactScannerOpen(final ObserverContext<RegionCoprocessorEnvironment> c,
      final Store store, List<? extends KeyValueScanner> scanners, final ScanType scanType,
      final long earliestPutTs, final InternalScanner s, CompactionRequest request); static final String THEMIS_DELETE_THEMIS_DELETED_DATA_WHEN_COMPACT; static final String SINGLE_ROW_PRIMARY_QUALIFIER; static final String LOCK_FAMILY_DELETE; }
@Test public void testPreFlushScannerOpen() throws Exception { if (TEST_UTIL != null) { ZooKeeperWatcher zk = new ZooKeeperWatcher(conf, "test", null, true); HBaseAdmin admin = new HBaseAdmin(connection); prewriteTestDataForPreFlushAndPreCompact(); ZKUtil.deleteNodeFailSilent(zk, ThemisMasterObserver.getThemisExpiredTsZNodePath(zk)); admin.flush(TABLENAME); Assert.assertEquals(6, getRowByScan().size()); deleteOldDataAndUpdateTs(); prewriteTestDataForPreFlushAndPreCompact(); ZKUtil.createSetData(zk, ThemisMasterObserver.getThemisExpiredTsZNodePath(zk), Bytes.toBytes(String.valueOf(Long.MIN_VALUE))); admin.flush(TABLENAME); Assert.assertEquals(6, getRowByScan().size()); deleteOldDataAndUpdateTs(); prewriteTestDataForPreFlushAndPreCompact(); ZKUtil.createSetData(zk, ThemisMasterObserver.getThemisExpiredTsZNodePath(zk), Bytes.toBytes(String.valueOf(prewriteTs + 1))); admin.flush(TABLENAME); Assert.assertEquals(6, getRowByScan().size()); deleteOldDataAndUpdateTs(); prewriteTestDataForPreFlushAndPreCompact(); ZKUtil.createSetData(zk, ThemisMasterObserver.getThemisExpiredTsZNodePath(zk), Bytes.toBytes(String.valueOf(prewriteTs + 5))); admin.flush(TABLENAME); Result result = getRowByScan(); Assert.assertEquals(3, result.size()); Assert.assertNotNull(result.getValue(FAMILY, QUALIFIER)); Column deleteColumn = ColumnUtil.getDeleteColumn(COLUMN); Assert.assertNotNull(result.getValue(deleteColumn.getFamily(), deleteColumn.getQualifier())); Column putColumn = ColumnUtil.getPutColumn(COLUMN); Assert.assertNotNull(result.getValue(putColumn.getFamily(), putColumn.getQualifier())); deleteOldDataAndUpdateTs(); admin.close(); zk.close(); } }