docs: void onConnect() throws InterruptedException, IOException
      {
          if (state.get() != NodeState.Fresh) {
              if (previousZKSessionStillActive()) {
                  LOG.info("ZooKeeper session re-established before timeout.");
                  return;
              }
              LOG.warn("Rejoined after session timeout. Forcing shutdown and clean startup.");
              ensureCleanStartup();
          }
  
          LOG.info("Connected to Zookeeper (ID: {}).", myNodeID);
          ZKUtils.ensureOrdasityPaths(zk, name, config.workUnitName, config.workUnitShortName);
          joinCluster();
          listener.onJoin(zk);
          if (watchesRegistered.compareAndSet(false, true)) {
              registerWatchers();
          }
          initialized.set(true);
          initializedLatch.countDown();
          setState(NodeState.Started);
          claimer.requestClaim();
          verifyIntegrity();
  
          balancingPolicy.onConnect();
          if (config.enableAutoRebalance) {
              scheduleRebalancing();
          }
      }