class 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);
    }
  }

}