docs: private void registerWatchers() throws InterruptedException
      {
          ClusterNodesChangedListener nodesChangedListener = new ClusterNodesChangedListener(this);
          VerifyIntegrityListener<String> verifyIntegrityListener =
                  new VerifyIntegrityListener<String>(this);
          ZKDeserializers.StringDeserializer stringDeser = new ZKDeserializers.StringDeserializer();
  
          try {
              nodes = ZooKeeperMap.create(zk, String.format("/%s/nodes", name),
                      new ZKDeserializers.NodeInfoDeserializer(), nodesChangedListener);
              allWorkUnits = ZooKeeperMap.create(zk, String.format("/%s", config.workUnitName),
                      new ZKDeserializers.ObjectNodeDeserializer(), new VerifyIntegrityListener<ObjectNode>(this));
  
              workUnitMap = ZooKeeperMap.create(zk, String.format("/%s/claimed-%s", name, config.workUnitShortName),
                      stringDeser, verifyIntegrityListener);
  
              // Watch handoff requests and results.
              if (config.useSoftHandoff) {
                  handoffRequests = ZooKeeperMap.create(zk, String.format("/%s/handoff-requests", name),
                          stringDeser, verifyIntegrityListener);
  
                  handoffResults = ZooKeeperMap.create(zk, String.format("/%s/handoff-result", name),
                          stringDeser, handoffResultsListener);
              } else {
                  handoffRequests = new HashMap<String, String>();
                  handoffResults = new HashMap<String, String>();
              }
  
              // If smart balancing is enabled, watch for changes to the cluster's workload.
              if (config.useSmartBalancing) {
                  loadMap = ZooKeeperMap.<Double>create(zk, String.format("/%s/meta/workload", name),
                          new ZKDeserializers.DoubleDeserializer());
              }
          } catch (KeeperException e) {
              throw ZKException.from(e);
          } catch (ZooKeeperConnectionException e) {
              throw ZKException.from(e);
          }
      }