org.apache.hadoop.hbase.zookeeper.RecoverableZooKeeper#getData ( )源码实例Demo

public static boolean shouldCaptureIncrementalChanges(FileSystem fs,Path rootDir) throws StandardException{
    boolean shouldRegister = false;
    try {
        boolean enabled = incrementalBackupEnabled();
        if (enabled) {
            RecoverableZooKeeper zooKeeper = ZkUtils.getRecoverableZooKeeper();
            String spliceBackupPath = BackupUtils.getBackupPath();
            if (zooKeeper.exists(spliceBackupPath, false)==null){
                return false;
            }
            boolean isRestoreMode = SIDriver.driver().lifecycleManager().isRestoreMode();
            if (!isRestoreMode) {
                if (BackupUtils.existsDatabaseBackup(fs, rootDir)) {
                    if (LOG.isDebugEnabled()) {
                        SpliceLogUtils.debug(LOG, "There exists a successful full or incremental backup in the system");
                    }
                    shouldRegister = true;
                } else {
                    List<String> backupJobs = zooKeeper.getChildren(spliceBackupPath, false);
                    for (String backupId : backupJobs) {
                        String path = spliceBackupPath + "/" + backupId;
                        byte[] data = zooKeeper.getData(path, false, null);
                        BackupJobStatus status = BackupJobStatus.parseFrom(data);
                        if (status.getScope() == BackupJobStatus.BackupScope.DATABASE) {
                            if (LOG.isDebugEnabled()) {
                                SpliceLogUtils.debug(LOG, "A database backup is running");
                            }
                            shouldRegister = true;
                        }
                    }
                }
            }
        }
        return shouldRegister;
    }
    catch (Exception e) {
        e.printStackTrace();
        throw StandardException.plainWrapException(e);
    }
}
 

private long getMasterTimestamp() throws IOException, KeeperException, InterruptedException {
    String masterClusterKey = getClusterKey(masterCluster);
    Configuration conf = ReplicationUtils.createConfiguration(masterClusterKey);

    try (ZKWatcher masterZkw = new ZKWatcher(conf, "replication monitor", null, false)) {
        RecoverableZooKeeper rzk = masterZkw.getRecoverableZooKeeper();
        String rootNode = HConfiguration.getConfiguration().getSpliceRootPath();
        String node = rootNode + HConfiguration.MAX_RESERVED_TIMESTAMP_PATH;
        byte[] data = rzk.getData(node, false, null);
        long ts = Bytes.toLong(data);
        return ts;
    }
}
 

private void publishServer(RecoverableZooKeeper rzk, String hostname, int port) throws InterruptedException, KeeperException {
    try {
        HostAndPort hostAndPort = HostAndPort.fromParts(hostname, port);
        queueZkPath = rzk.getZooKeeper().create(queueZkPath, Bytes.toBytes(hostAndPort.toString()),
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        rzk.getData(queueZkPath, new QueueWatcher(), null);
    } catch (Exception e) {
        LOG.error("Couldn't register OlapServer due to unexpected exception", e);
        throw e;
    }
}