protected void connectionEvent(WatchedEvent event) {
        switch (event.getState()) {
            case SyncConnected:
                // Now, this callback can be invoked before the this.zookeeper is
                // set.
                // Wait a little while.
                long finished = System.currentTimeMillis() + this.conf.getZkSyncTimeMs();
                while (System.currentTimeMillis() < finished) {
                    ThreadUtils.sleep(1);
                    if (this.recoverableZooKeeper != null) {
                        break;
                    }
                }
                if (this.recoverableZooKeeper == null) {
                    logger.error("ZK is null on connection event -- see stack trace "
                            + "for the stack trace when constructor was called on this zkw",
                            this.constructorCaller);
                    throw new NullPointerException("ZK is null");
                }
                // this.identifier = this.identifier + "-0x" +
                // Long.toHexString(this.recoverableZooKeeper.getSessionId());
                // Update our identifier. Otherwise ignore.
                if (logger.isDebugEnabled()) {
                    logger.debug(Long.toHexString(this.recoverableZooKeeper.getSessionId()) + " connected");
                }
                break;

            case SaslAuthenticated:
                if (ZKUtil.isSecureZooKeeper()) {
                    // We are authenticated, clients can proceed.
                    saslLatch.countDown();
                }
                break;

            case AuthFailed:
                if (ZKUtil.isSecureZooKeeper()) {
                    // We could not be authenticated, but clients should proceed
                    // anyway.
                    // Only access to znodes that require SASL authentication will
                    // be
                    // denied. The client may never need to access them.
                    saslLatch.countDown();
                }
                break;

            // Abort or other the server if Disconnected or Expired
            case Disconnected:
                if (logger.isDebugEnabled()) {
                    logger.debug(prefix("Received Disconnected from ZooKeeper, ignoring"));
                }
                recoverableZooKeeper.ensureConnectivity(null);
                break;
            case Expired:
                if (ZKUtil.isSecureZooKeeper()) {
                    // We consider Expired equivalent to AuthFailed for this
                    // connection. Authentication is never going to complete. The
                    // client should proceed to do cleanup.
                    saslLatch.countDown();
                }
                String msg =
                        prefix(Long.toHexString(this.recoverableZooKeeper.getSessionId())
                                + " received expired from " + "ZooKeeper, aborting");
                // TODO: One thought is to add call to ZooKeeperListener so say,
                // ZooKeeperNodeTracker can zero out its data values.
                if (this.abortable != null) {
                    this.abortable.abort(msg, new KeeperException.SessionExpiredException());
                }

                recoverableZooKeeper.ensureConnectivity(null);
                break;
        }
    }
Called when there is a connection-related event via the Watcher callback.
<p/>
If Disconnected or Expired, this should shutdown the cluster. But, since we send a
KeeperException.SessionExpiredException along with the abort call, it's possible for the
Abortable to catch it and try to create a new session with ZooKeeper. This is what the client
does in HCM.
<p/>
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/common/zookeeper/ZooKeeperWatcher.java