java/openmldb-taskmanager/src/main/java/com/_4paradigm/openmldb/taskmanager/zk/FailoverWatcher.java
public boolean blockUntilActive() {
    while (true) {
      try {
        if (ZooKeeperUtil.createEphemeralNodeAndWatch(this, masterZnode, hostPort.getHostPort()
            .getBytes())) {

          // We are the master, return
          hasActiveServer.set(true);
          becomeActiveServer.set(true);
          LOG.info("Become active master in " + hostPort.getHostPort());
          return true;
        }

        hasActiveServer.set(true);

        // we start the server with the same ip_port stored in master znode, that means we want to
        // restart the server?
        String msg;
        byte[] bytes = ZooKeeperUtil.getDataAndWatch(this, masterZnode);
        if (bytes == null) {
          msg = ("A master was detected, but went down before its address "
              + "could be read.  Attempting to become the next active master");
        } else {
          if (hostPort.getHostPort().equals(new String(bytes))) {
            msg = ("Current master has this master's address, " + hostPort.getHostPort() + "; master was restarted? Deleting node.");
            // Hurry along the expiration of the znode.
            ZooKeeperUtil.deleteNode(this, masterZnode);
          } else {
            msg = "Another master " + new String(bytes) + " is the active master, "
                + hostPort.getHostPort() + "; waiting to become the next active master";
          }
        }
        LOG.info(msg);
      } catch (KeeperException ke) {
        LOG.error("Received an unexpected KeeperException when block to become active, aborting",
          ke);
        return false;
      }

      synchronized (hasActiveServer) {
        while (hasActiveServer.get()) {
          try {
            hasActiveServer.wait();
          } catch (InterruptedException e) {
            // We expect to be interrupted when a master dies, will fall out if so
            if (LOG.isDebugEnabled()) {
              LOG.debug("Interrupted while waiting to be master");
            }
            return false;
          }
        }
      }
    }
  }
/**
   * Implement the logic of server to wait to become active master.
   *
   * @return false if error to wait to become active master
   */
Implement the logic of server to wait to become active master.
@return false if error to wait to become active master
Implement the logic of server to wait to become active master . @return false if error to wait to become active master
public boolean blockUntilActive() {
    while (true) {
      try {
        if (ZooKeeperUtil.createEphemeralNodeAndWatch(this, masterZnode, hostPort.getHostPort()
            .getBytes())) {
          hasActiveServer.set(true);
          becomeActiveServer.set(true);
          LOG.info("Become active master in " + hostPort.getHostPort());
          return true;
        }
        hasActiveServer.set(true);
        String msg;
        byte[] bytes = ZooKeeperUtil.getDataAndWatch(this, masterZnode);
        if (bytes == null) {
          msg = ("A master was detected, but went down before its address "
              + "could be read.  Attempting to become the next active master");
        } else {
          if (hostPort.getHostPort().equals(new String(bytes))) {
            msg = ("Current master has this master's address, " + hostPort.getHostPort() + "; master was restarted? Deleting node.");
            ZooKeeperUtil.deleteNode(this, masterZnode);
          } else {
            msg = "Another master " + new String(bytes) + " is the active master, "
                + hostPort.getHostPort() + "; waiting to become the next active master";
          }
        }
        LOG.info(msg);
      } catch (KeeperException ke) {
        LOG.error("Received an unexpected KeeperException when block to become active, aborting",
          ke);
        return false;
      }
      synchronized (hasActiveServer) {
        while (hasActiveServer.get()) {
          try {
            hasActiveServer.wait();
          } catch (InterruptedException e) {
            if (LOG.isDebugEnabled()) {
              LOG.debug("Interrupted while waiting to be master");
            }
            return false;
          }
        }
      }
    }
  }
public boolean blockUntilActive ( ) { while ( true ) { try { if ( ZooKeeperUtil . createEphemeralNodeAndWatch ( this , masterZnode , hostPort . getHostPort ( ) . getBytes ( ) ) ) { hasActiveServer . set ( true ) ; becomeActiveServer . set ( true ) ; LOG . info ( "Become active master in " + hostPort . getHostPort ( ) ) ; return true ; } hasActiveServer . set ( true ) ; String msg ; byte [ ] bytes = ZooKeeperUtil . getDataAndWatch ( this , masterZnode ) ; if ( bytes == null ) { msg = ( "A master was detected, but went down before its address " + "could be read.  Attempting to become the next active master" ) ; } else { if ( hostPort . getHostPort ( ) . equals ( new String ( bytes ) ) ) { msg = ( "Current master has this master's address, " + hostPort . getHostPort ( ) + "; master was restarted? Deleting node." ) ; ZooKeeperUtil . deleteNode ( this , masterZnode ) ; } else { msg = "Another master " + new String ( bytes ) + " is the active master, " + hostPort . getHostPort ( ) + "; waiting to become the next active master" ; } } LOG . info ( msg ) ; } catch ( KeeperException ke ) { LOG . error ( "Received an unexpected KeeperException when block to become active, aborting" , ke ) ; return false ; } synchronized ( hasActiveServer ) { while ( hasActiveServer . get ( ) ) { try { hasActiveServer . wait ( ) ; } catch ( InterruptedException e ) { if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "Interrupted while waiting to be master" ) ; } return false ; } } } } }
Implement the logic of server to wait to become active master.
Implement the logic of server to wait to become active master .
// We are the master, return // we start the server with the same ip_port stored in master znode, that means we want to // restart the server? // Hurry along the expiration of the znode. // We expect to be interrupted when a master dies, will fall out if so