Java 类org.apache.hadoop.hbase.zookeeper.RecoverableZooKeeper 实例源码

private void testZNodeACLs() throws IOException, KeeperException, InterruptedException {

    ZooKeeperWatcher watcher = new ZooKeeperWatcher(conf, "IntegrationTestZnodeACLs", null);
    RecoverableZooKeeper zk = ZKUtil.connect(this.conf, watcher);

    String baseZNode = watcher.baseZNode;

    LOG.info("");
    LOG.info("***********************************************************************************");
    LOG.info("Checking ZK permissions, root znode: " + baseZNode);
    LOG.info("***********************************************************************************");
    LOG.info("");

    checkZnodePermsRecursive(watcher, zk, baseZNode);

    LOG.info("Checking ZK permissions: SUCCESS");
  }

private void checkZnodePermsRecursive(ZooKeeperWatcher watcher,
    RecoverableZooKeeper zk, String znode) throws KeeperException, InterruptedException {

  boolean expectedWorldReadable = watcher.isClientReadable(znode);

  assertZnodePerms(zk, znode, expectedWorldReadable);

  try {
    List<String> children = zk.getChildren(znode, false);

    for (String child : children) {
      checkZnodePermsRecursive(watcher, zk, ZKUtil.joinZNode(znode, child));
    }
  } catch (KeeperException ke) {
    // if we are not authenticated for listChildren, it is fine.
    if (ke.code() != Code.NOAUTH) {
      throw ke;
    }
  }
}

/**
 * Ensure Armeria's dependencies do not cause a trouble with hbase-shaded-client.
 *
 * @see <a href="https://issues.apache.org/jira/browse/HBASE-14963">HBASE-14963</a>
 */
@Test(expected = NotAllMetaRegionsOnlineException.class)
public void testGuavaConflict() throws Exception {
    // Make sure Armeria is available in the class path.
    assertThat(Version.identify(Server.class.getClassLoader())).isNotNull();
    // Make sure newer Guava is available in the class path.
    assertThat(Stopwatch.class.getDeclaredConstructor().getModifiers()).is(new Condition<>(
            value -> !Modifier.isPublic(value),
            "Recent Guava Stopwatch should have non-public default constructor."));

    final MetaTableLocator locator = new MetaTableLocator();
    final ZooKeeperWatcher zkw = mock(ZooKeeperWatcher.class);
    final RecoverableZooKeeper zk = mock(RecoverableZooKeeper.class);
    when(zkw.getRecoverableZooKeeper()).thenReturn(zk);
    when(zk.exists(any(), any())).thenReturn(new Stat(0, 0, 0, 0, 1, 1, 1, 0, 0, 0, 0));

    locator.waitMetaRegionLocation(zkw, 100);
}

private void testZNodeACLs() throws IOException, KeeperException, InterruptedException {

    ZKWatcher watcher = new ZKWatcher(conf, "IntegrationTestZnodeACLs", null);
    RecoverableZooKeeper zk = ZKUtil.connect(this.conf, watcher);

    String baseZNode = watcher.znodePaths.baseZNode;

    LOG.info("");
    LOG.info("***********************************************************************************");
    LOG.info("Checking ZK permissions, root znode: " + baseZNode);
    LOG.info("***********************************************************************************");
    LOG.info("");

    checkZnodePermsRecursive(watcher, zk, baseZNode);

    LOG.info("Checking ZK permissions: SUCCESS");
  }

private void checkZnodePermsRecursive(ZKWatcher watcher,
    RecoverableZooKeeper zk, String znode) throws KeeperException, InterruptedException {

  boolean expectedWorldReadable = watcher.znodePaths.isClientReadable(znode);

  assertZnodePerms(zk, znode, expectedWorldReadable);

  try {
    List<String> children = zk.getChildren(znode, false);

    for (String child : children) {
      checkZnodePermsRecursive(watcher, zk, ZNodePaths.joinZNode(znode, child));
    }
  } catch (KeeperException ke) {
    // if we are not authenticated for listChildren, it is fine.
    if (ke.code() != Code.NOAUTH && ke.code() != Code.NONODE) {
      throw ke;
    }
  }
}

/**
 * Deletes just the splice-specific paths in zookeeper.  Does not delete hbase paths.
 */
public static void cleanZookeeper() throws InterruptedException, KeeperException{
    RecoverableZooKeeper rzk=getRecoverableZooKeeper();
    String rootPath=HConfiguration.getConfiguration().getSpliceRootPath();
    for(String path : HConfiguration.zookeeperPaths){
        path=rootPath+path;
        if(rzk.exists(path,false)!=null){
            for(String child : rzk.getChildren(path,false)){
                for(String grandChild : rzk.getChildren(path+"/"+child,false)){
                    rzk.delete(path+"/"+child+"/"+grandChild,-1);
                }
                rzk.delete(path+"/"+child,-1);
            }
            rzk.delete(path,-1);
        }
    }
}

private void assertZnodePerms(RecoverableZooKeeper zk, String znode,
    boolean expectedWorldReadable) throws KeeperException, InterruptedException {
  Stat stat = new Stat();
  List<ACL> acls = zk.getZooKeeper().getACL(znode, stat);
  String[] superUsers = superUser == null ? null : superUser.split(",");

  LOG.info("Checking ACLs for znode znode:" + znode + " acls:" + acls);

  for (ACL acl : acls) {
    int perms = acl.getPerms();
    Id id = acl.getId();
    // We should only set at most 3 possible ACL for 3 Ids. One for everyone, one for superuser
    // and one for the hbase user
    if (Ids.ANYONE_ID_UNSAFE.equals(id)) {
      // everyone should be set only if we are expecting this znode to be world readable
      assertTrue(expectedWorldReadable);
      // assert that anyone can only read
      assertEquals(perms, Perms.READ);
    } else if (superUsers != null && ZooKeeperWatcher.isSuperUserId(superUsers, id)) {
      // assert that super user has all the permissions
      assertEquals(perms, Perms.ALL);
    } else if (new Id("sasl", masterPrincipal).equals(id)) {
      // hbase.master.kerberos.principal?
      assertEquals(perms, Perms.ALL);
    } else {
      fail("An ACL is found which is not expected for the znode:" + znode + " , ACL:" + acl);
    }
  }
}

/**
 * Gets a direct interface to a ZooKeeper instance.
 *
 * @return a direct interface to ZooKeeper.
 */
public static RecoverableZooKeeper getRecoverableZooKeeper(){
    try{
        return zkManager.getRecoverableZooKeeper();
    }catch(ZooKeeperConnectionException e){
        LOG.error("Unable to connect to zookeeper, aborting",e);
        throw new RuntimeException(e);
    }
}

public static boolean recursiveSafeCreate(String path,byte[] bytes,List<ACL> acls,CreateMode createMode) throws InterruptedException, KeeperException{
    if(path==null || path.length()<=0) return true; //nothing to do, we've gone all the way to the root
    RecoverableZooKeeper rzk=getRecoverableZooKeeper();
    try{
        return safeCreate(path,bytes,acls,createMode,rzk);
    }catch(KeeperException e){
        if(e.code()==KeeperException.Code.NONODE){
            //parent node doesn't exist, so recursively create it, and then try and create your node again
            String parent=path.substring(0,path.lastIndexOf('/'));
            recursiveSafeCreate(parent,new byte[]{},acls,CreateMode.PERSISTENT);
            return safeCreate(path,bytes,acls,createMode);
        }else
            throw e;
    }
}

public static boolean safeCreate(String path,byte[] bytes,List<ACL> acls,CreateMode createMode,RecoverableZooKeeper zooKeeper) throws KeeperException, InterruptedException{
    try{
        zooKeeper.create(path,bytes,acls,createMode);
        return true;
    }catch(KeeperException ke){
        if(ke.code()!=KeeperException.Code.NODEEXISTS)
            throw ke;
        else
            return true;
    }
}

public static boolean validZookeeper() throws InterruptedException, KeeperException{
    RecoverableZooKeeper rzk=getRecoverableZooKeeper();

    String rootPath=HConfiguration.getConfiguration().getSpliceRootPath();
    for(String path : HConfiguration.zookeeperPaths){
        if(rzk.exists(rootPath+path,false)==null)
            return false;
    }
    return true;
}

public static HBaseSIEnvironment loadEnvironment(Clock clock,RecoverableZooKeeper rzk) throws IOException{
    HBaseSIEnvironment env = INSTANCE;
    if(env==null){
        synchronized(HBaseSIEnvironment.class){
            env = INSTANCE;
            if(env==null){
                env = INSTANCE = new HBaseSIEnvironment(rzk,clock);
            }
        }
    }
    return env;
}

@SuppressWarnings("unchecked")
public HBaseSIEnvironment(RecoverableZooKeeper rzk,Clock clock) throws IOException{
    ByteComparisons.setComparator(HBaseComparator.INSTANCE);
    this.config=HConfiguration.getConfiguration();

    this.timestampSource =new ZkTimestampSource(config,rzk);
    this.partitionCache = PartitionCacheService.loadPartitionCache(config);
    this.partitionFactory =TableFactoryService.loadTableFactory(clock, this.config,partitionCache);
    TxnNetworkLayerFactory txnNetworkLayerFactory= TableFactoryService.loadTxnNetworkLayer(this.config);
    this.txnStore = new CoprocessorTxnStore(txnNetworkLayerFactory,timestampSource,null);
    int completedTxnCacheSize = config.getCompletedTxnCacheSize();
    int completedTxnConcurrency = config.getCompletedTxnConcurrency();
    this.txnSupplier = new CompletedTxnCacheSupplier(txnStore,completedTxnCacheSize,completedTxnConcurrency);
    this.txnStore.setCache(txnSupplier);
    this.opFactory =HOperationFactory.INSTANCE;
    this.txnOpFactory = new SimpleTxnOperationFactory(exceptionFactory(),opFactory);
    this.clock = clock;
    this.fileSystem =new HNIOFileSystem(FileSystem.get((Configuration) config.getConfigSource().unwrapDelegate()), exceptionFactory());
    this.snowflakeFactory = new HSnowflakeFactory();
    this.clusterHealthFactory = new HClusterHealthFactory(rzk);
    this.ignoreTxnSupplier = new IgnoreTxnSupplier(partitionFactory, txnOpFactory);

    this.keepAlive = new QueuedKeepAliveScheduler(config.getTransactionKeepAliveInterval(),
            config.getTransactionTimeout(),
            config.getTransactionKeepAliveThreads(),
            txnStore);
    siDriver = SIDriver.loadDriver(this);
}

/**
 * An HFile is eligible for incremental backup if
 * 1) There is an ongoing full backup, flush is not triggered by preparing and backup for this region is done.
 * 2) There is no ongoing backup, AND there is a previous full/incremental backup
 * 3) There is an ongoing incremental backup
 * @param fileName
 * @throws StandardException
 */
public static void captureIncrementalChanges( Configuration conf,
                                              HRegion region,
                                              String path,
                                              FileSystem fs,
                                              Path rootDir,
                                              Path backupDir,
                                              String tableName,
                                              String fileName,
                                              boolean preparing) throws StandardException {
    boolean shouldRegister = false;
    try {
        RecoverableZooKeeper zooKeeper = ZkUtils.getRecoverableZooKeeper();
        String spliceBackupPath = HConfiguration.getConfiguration().getBackupPath();
        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 if (zooKeeper.exists(spliceBackupPath, false) != null) {

            if (LOG.isDebugEnabled()) {
                SpliceLogUtils.debug(LOG, "A backup is running");
            }
            shouldRegister = true;
        }

        if (shouldRegister) {
            registerHFile(conf, fs, backupDir, region, fileName);
        }
    }
    catch (Exception e) {
        e.printStackTrace();
        throw Exceptions.parseException(e);
    }
}

private void publishServer(RecoverableZooKeeper rzk, ServerName serverName, String hostname, int port) throws InterruptedException, KeeperException {
    String root = HConfiguration.getConfiguration().getSpliceRootPath();

    try {
        HostAndPort hostAndPort = HostAndPort.fromParts(hostname, port);
        masterPath = root + HBaseConfiguration.OLAP_SERVER_PATH + "/" + serverName;
        rzk.create(masterPath, Bytes.toBytes(hostAndPort.toString()), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        rzk.getData(masterPath, this, null);
    } catch (Exception e) {
        LOG.error("Couldn't register OlapServer due to unexpected exception", e);
        throw e;
    }
}

@Override
public RecoverableZooKeeper getRecoverableZooKeeper() {
  return zk;
}

@Override
public RecoverableZooKeeper getRecoverableZooKeeper() {
  return zk;
}

private void assertZnodePerms(RecoverableZooKeeper zk, String znode,
    boolean expectedWorldReadable) throws KeeperException, InterruptedException {
  Stat stat = new Stat();
  List<ACL> acls;
  try {
    acls = zk.getZooKeeper().getACL(znode, stat);
  } catch (NoNodeException ex) {
    LOG.debug("Caught exception for missing znode", ex);
    // the znode is deleted. Probably it was a temporary znode (like RIT).
    return;
  }
  String[] superUsers = superUser == null ? null : superUser.split(",");

  LOG.info("Checking ACLs for znode znode:" + znode + " acls:" + acls);

  for (ACL acl : acls) {
    int perms = acl.getPerms();
    Id id = acl.getId();
    // We should only set at most 3 possible ACL for 3 Ids. One for everyone, one for superuser
    // and one for the hbase user
    if (Ids.ANYONE_ID_UNSAFE.equals(id)) {
      // everyone should be set only if we are expecting this znode to be world readable
      assertTrue(expectedWorldReadable);
      // assert that anyone can only read
      assertEquals(perms, Perms.READ);
    } else if (superUsers != null && ZKWatcher.isSuperUserId(superUsers, id)) {
      // assert that super user has all the permissions
      assertEquals(perms, Perms.ALL);
    } else if (new Id("sasl", masterPrincipal).equals(id)) {
      // hbase.master.kerberos.principal?
      assertEquals(perms, Perms.ALL);
    } else {
      fail("An ACL is found which is not expected for the znode:" + znode + " , ACL:" + acl);
    }
  }
}

public static void delete(String path) throws InterruptedException, KeeperException{
    RecoverableZooKeeper rzk=getRecoverableZooKeeper();
    rzk.delete(path,-1);
}

public static boolean isSpliceLoaded() throws InterruptedException, KeeperException{
    RecoverableZooKeeper rzk=getRecoverableZooKeeper();
    String path=HConfiguration.getConfiguration().getSpliceRootPath()+HConfiguration.STARTUP_PATH;
    return rzk.exists(path,false)!=null;
}

public static void spliceFinishedLoading() throws InterruptedException, KeeperException{
    RecoverableZooKeeper rzk=getRecoverableZooKeeper();
    String path=HConfiguration.getConfiguration().getSpliceRootPath()+HConfiguration.STARTUP_PATH;
    rzk.create(path,Bytes.toBytes(0L),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
}

public RecoverableZooKeeper getRecoverableZooKeeper() throws ZooKeeperConnectionException{
    return rzk;
}

public <T> T executeUnlessExpired(Command<T> command) throws InterruptedException, KeeperException{
    /*
     * What actually happens is that, in the event of a long network partition, ZooKeeper will throw
     * ConnectionLoss exceptions, but it will NOT throw a SessionExpired exception until it reconnects, even
     * if it's been disconnected for CLEARLY longer than the session timeout.
     *
     * To deal with this, we have to basically loop through our command repeatedly until we either
     *
     * 1. Succeed.
     * 2. Get a SessionExpired event from ZooKeeper
     * 3. Spent more than 2*sessionTimeout ms attempting the request
     * 4. Get some other kind of Zk error (NoNode, etc).
     */
    RecoverableZooKeeper rzk;
    try{
        rzk=getRecoverableZooKeeper();
    }catch(ZooKeeperConnectionException e){
        throw new KeeperException.SessionExpiredException();
    }
    //multiple by 2 to make absolutely certain we're timed out.
    int sessionTimeout=2*rzk.getZooKeeper().getSessionTimeout();
    long nextTime=System.currentTimeMillis();
    long startTime=System.currentTimeMillis();
    while((int)(nextTime-startTime)<sessionTimeout){
        try{
            return command.execute(rzk);
        }catch(KeeperException ke){
            switch(ke.code()){
                case CONNECTIONLOSS:
                case OPERATIONTIMEOUT:
                    LOG.warn("Detected a Connection issue("+ke.code()+") with ZooKeeper, retrying");
                    nextTime=System.currentTimeMillis();
                    break;
                default:
                    throw ke;
            }
        }
    }

    //we've run out of time, our session has almost certainly expired. Give up and explode
    throw new KeeperException.SessionExpiredException();
}

public ZkTimestampSource(SConfiguration config,RecoverableZooKeeper rzk) {
    _rzk = rzk;
    initialize(config);
}

/**
 * @param rzk the ZooKeeper node to base off
 * @param blockNode Pointer to the specific znode instance that is specifically configured for timestamp block storage
 */
public ZkTimestampBlockManager(RecoverableZooKeeper rzk,String blockNode) {
    this.rzk = rzk;
    this.blockNode = blockNode;
}

public HClusterHealthFactory(RecoverableZooKeeper rzk){
    this.rzk = rzk;
}

HClusterHealthWatcher(RecoverableZooKeeper rzk) {
    this.rzk = rzk;
}

@Override
public void start(CoprocessorEnvironment ctx) throws IOException {
    try {
        LOG.info("Starting SpliceMasterObserver");

        LOG.info("Starting Timestamp Master Observer");

        ZooKeeperWatcher zkw = ((MasterCoprocessorEnvironment)ctx).getMasterServices().getZooKeeper();
        RecoverableZooKeeper rzk = zkw.getRecoverableZooKeeper();

        HBaseSIEnvironment env=HBaseSIEnvironment.loadEnvironment(new SystemClock(),rzk);
        SConfiguration configuration=env.configuration();

        String timestampReservedPath=configuration.getSpliceRootPath()+HConfiguration.MAX_RESERVED_TIMESTAMP_PATH;
        int timestampPort=configuration.getTimestampServerBindPort();
        int timestampBlockSize = configuration.getTimestampBlockSize();

        TimestampBlockManager tbm= new ZkTimestampBlockManager(rzk,timestampReservedPath);
        this.timestampServer =new TimestampServer(timestampPort,tbm,timestampBlockSize);

        this.timestampServer.startServer();

        if (!configuration.getOlapServerExternal()) {
            int olapPort = configuration.getOlapServerBindPort();
            this.olapServer = new OlapServer(olapPort, env.systemClock());
            this.olapServer.startServer(configuration);
        }

        /*
         * We create a new instance here rather than referring to the singleton because we have
         * a problem when booting the master and the region server in the same JVM; the singleton
         * then is unable to boot on the master side because the regionserver has already started it.
         *
         * Generally, this isn't a problem because the underlying singleton is constructed on demand, so we
         * will still only create a single manager per JVM in a production environment, and we avoid the deadlock
         * issue during testing
         */
        this.manager = new DatabaseLifecycleManager();
        super.start(ctx);
    } catch (Throwable t) {
        throw CoprocessorUtils.getIOException(t);
    }
}

public static boolean backupInProgress() throws Exception {
    String path = HConfiguration.getConfiguration().getBackupPath();
    RecoverableZooKeeper zooKeeper = ZkUtils.getRecoverableZooKeeper();
    Stat stat = zooKeeper.exists(path, false);
    return (stat != null);
}

public static boolean backupCanceled() throws KeeperException, InterruptedException {
    RecoverableZooKeeper zooKeeper = ZkUtils.getRecoverableZooKeeper();
    String path = HConfiguration.getConfiguration().getBackupPath();
    return zooKeeper.exists(path, false) == null;
}

public SpliceMasterLock(String parentPath, String path, RecoverableZooKeeper zooKeeper) {
    this.path = path;
    this.parent = parentPath;
    this.zk = zooKeeper;
    this.semaphore = new Semaphore(0);
}

T execute(RecoverableZooKeeper zooKeeper) throws InterruptedException, KeeperException;