ThemisMasterObserver extends BaseMasterObserver { public void cleanLockBeforeTimestamp(long ts) throws IOException { List<String> tableNames = getThemisTables(connection); for (String tableName : tableNames) { LOG.info("start to clean expired lock for themis table:" + tableName); byte[] tableNameBytes = Bytes.toBytes(tableName); HTableInterface hTable = null; int cleanedLockCount = 0; try { hTable = new HTable(connection.getConfiguration(), tableName); Scan scan = new Scan(); scan.addFamily(ColumnUtil.LOCK_FAMILY_NAME); scan.setTimeRange(0, ts); ResultScanner scanner = hTable.getScanner(scan); Result result = null; while ((result = scanner.next()) != null) { for (KeyValue kv : result.list()) { ThemisLock lock = ThemisLock.parseFromByte(kv.getValue()); Column dataColumn = ColumnUtil.getDataColumnFromConstructedQualifier(new Column(kv.getFamily(), kv.getQualifier())); lock.setColumn(new ColumnCoordinate(tableNameBytes, kv.getRow(), dataColumn.getFamily(), dataColumn.getQualifier())); lockCleaner.cleanLock(lock); ++cleanedLockCount; LOG.info("themis clean expired lock, lockTs=" + kv.getTimestamp() + ", expiredTs=" + ts + ", lock=" + ThemisLock.parseFromByte(kv.getValue())); } } scanner.close(); } finally { if (hTable != null) { hTable.close(); } } LOG.info("finish clean expired lock for themis table:" + tableName + ", cleanedLockCount=" + cleanedLockCount); } }  @Override void start(CoprocessorEnvironment ctx); @Override void stop(CoprocessorEnvironment ctx); @Override void preCreateTable(ObserverContext<MasterCoprocessorEnvironment> ctx,
      HTableDescriptor desc, HRegionInfo[] regions); @Override void postGetTableDescriptors(ObserverContext<MasterCoprocessorEnvironment> ctx,
      List<HTableDescriptor> descriptors); static boolean isThemisEnableFamily(HColumnDescriptor desc); static boolean isThemisEnableTable(HTableDescriptor desc); static List<String> getThemisTables(HConnection connection); static String getThemisExpiredTsZNodePath(ZooKeeperWatcher zk); static long getThemisExpiredTsFromZk(ZooKeeperWatcher zk); static long getThemisExpiredTsFromZk(ZooKeeperWatcher zk, String path); void setExpiredTsToZk(long currentExpiredTs); void cleanLockBeforeTimestamp(long ts); static final String RETURNED_THEMIS_TABLE_DESC; static final String THEMIS_ENABLE_KEY; static final String THEMIS_EXPIRED_TIMESTAMP_CALCULATE_PERIOD_KEY; static final String THEMIS_EXPIRED_DATA_CLEAN_ENABLE_KEY; static final String THEMIS_EXPIRED_TIMESTAMP_ZNODE_NAME; }
@Test public void testCleanTimeExpiredLock() throws IOException { ThemisMasterObserver masterObserver = new ThemisMasterObserver(); masterObserver.connection = connection; masterObserver.lockCleaner = new ServerLockCleaner(masterObserver.connection, new ThemisCoprocessorClient(connection)); writeLockAndData(COLUMN, prewriteTs); writeLockAndData(COLUMN_WITH_ANOTHER_TABLE, prewriteTs + 1); masterObserver.cleanLockBeforeTimestamp(prewriteTs); checkPrewriteColumnSuccess(COLUMN, prewriteTs); checkPrewriteColumnSuccess(COLUMN_WITH_ANOTHER_TABLE, prewriteTs + 1); masterObserver.cleanLockBeforeTimestamp(prewriteTs + 1); Assert.assertNull(readLockBytes(COLUMN, prewriteTs)); checkPrewriteColumnSuccess(COLUMN_WITH_ANOTHER_TABLE, prewriteTs + 1); writeLockAndData(COLUMN, prewriteTs + 1); masterObserver.cleanLockBeforeTimestamp(prewriteTs + 2); Assert.assertNull(readLockBytes(COLUMN, prewriteTs + 1)); Assert.assertNull(readLockBytes(COLUMN_WITH_ANOTHER_TABLE, prewriteTs + 1)); writeLockAndData(COLUMN, prewriteTs + 100); }