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

/**
 * Bump up timestamp if the provided timestamp value is larger than current timetamp
 * @param timestamp
 * @throws IOException
 * @throws KeeperException
 * @throws InterruptedException
 */
public static void setTimestamp(long timestamp) throws IOException, KeeperException, InterruptedException {
    TimestampSource timestampSource = SIDriver.driver().getTimestampSource();
    long currentTimestamp = timestampSource.currentTimestamp();
    if (currentTimestamp < timestamp) {
        RecoverableZooKeeper rzk = ZkUtils.getRecoverableZooKeeper();
        HBaseSIEnvironment env = HBaseSIEnvironment.loadEnvironment(new SystemClock(), rzk);
        ConfigurationSource configurationSource = env.configuration().getConfigSource();
        String rootNode = configurationSource.getString(HConfiguration.SPLICE_ROOT_PATH, HConfiguration.DEFAULT_ROOT_PATH);
        String node = rootNode + HConfiguration.MAX_RESERVED_TIMESTAMP_PATH;
        //if (LOG.isDebugEnabled()) {
        SpliceLogUtils.info(LOG, "bump up timestamp to %d", timestamp);
        //}
        byte[] data = Bytes.toBytes(timestamp);
        rzk.setData(node, data, -1 /* version */);
        timestampSource.refresh();
    }
    else {
        //if (LOG.isDebugEnabled()) {
        SpliceLogUtils.info(LOG, "current timestamp = %d >  %d",
                currentTimestamp, timestamp);
        //}
    }
}
 

public static void disableMaster(String masterClusterKey) throws InterruptedException, KeeperException, IOException {

        // Delete all peers from master cluster
        Configuration conf = ReplicationUtils.createConfiguration(masterClusterKey);
        ZKWatcher masterZkw = new ZKWatcher(conf, "replication monitor", null, false);
        RecoverableZooKeeper masterRzk = masterZkw.getRecoverableZooKeeper();
        String[] s = masterClusterKey.split(":");
        String hbaseRootDir = s[2];
        String peerPath = hbaseRootDir+"/replication/peers";
        List<String> peers = masterRzk.getChildren(peerPath, false);
        for (String peer : peers) {
            String p = peerPath + "/" + peer;
            List<String> children = masterRzk.getChildren(p, false);
            String peerStatePath = p + "/" + children.get(0);
            masterRzk.setData(peerStatePath, toByteArray(ReplicationProtos.ReplicationState.State.DISABLED), -1);
            System.out.println("Disabled peer " + peer);
        }
    }
 

private void reportDiagnostics(String diagnostics) {
    try {
        RecoverableZooKeeper rzk = ZkUtils.getRecoverableZooKeeper();
        String root = HConfiguration.getConfiguration().getSpliceRootPath();

        String diagnosticsPath = root + HBaseConfiguration.OLAP_SERVER_PATH + HBaseConfiguration.OLAP_SERVER_DIAGNOSTICS_PATH + "/" + queueName;

        if (rzk.exists(diagnosticsPath, false) != null) {
            rzk.setData(diagnosticsPath, Bytes.toBytes(diagnostics), -1);
        } else {
            rzk.create(diagnosticsPath, Bytes.toBytes(diagnostics), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    } catch (Exception e) {
        LOG.error("Exception while trying to report diagnostics", e);
        // ignore this exception during error reporting
    }
}
 

private void reportDiagnostics(String diagnostics) {
    try {
        RecoverableZooKeeper rzk = ZkUtils.getRecoverableZooKeeper();
        String root = HConfiguration.getConfiguration().getSpliceRootPath();

        String diagnosticsRoot = root + HBaseConfiguration.OLAP_SERVER_PATH + HBaseConfiguration.OLAP_SERVER_DIAGNOSTICS_PATH;
        zkSafeCreate(diagnosticsRoot);
        String diagnosticsPath = diagnosticsRoot + "/spark-" + queueName;

        if (rzk.exists(diagnosticsPath, false) != null) {
            rzk.setData(diagnosticsPath, com.splicemachine.primitives.Bytes.toBytes(diagnostics), -1);
        } else {
            rzk.create(diagnosticsPath, com.splicemachine.primitives.Bytes.toBytes(diagnostics), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    } catch (Exception e) {
        LOG.error("Exception while trying to report diagnostics", e);
        // ignore this exception during error reporting
    }
}
 

public static void setData(String path,byte[] data,int version, RecoverableZooKeeper rzk) throws IOException{
    try{
        rzk.setData(path,data,version);
    }catch(KeeperException | InterruptedException e){
        throw new IOException(e);
    }
}
 

private void setNewMasterTimestamp(String newMasterClusterKey,
                                   long ts) throws IOException, KeeperException, InterruptedException{
    Configuration conf = ReplicationUtils.createConfiguration(newMasterClusterKey);

    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;
        rzk.setData(node, Bytes.toBytes(ts), -1);
    }
}