code: public CuratorFramework mkClient(Map conf, List<String> servers, Object port,
                                       String root, final WatcherCallBack watcher) {
  
          CuratorFramework fk = Utils.newCurator(conf, servers, port, root);
  
          fk.getCuratorListenable().addListener(new CuratorListener() {
              @Override
              public void eventReceived(CuratorFramework _fk, CuratorEvent e) throws Exception {
                  if (e.getType().equals(CuratorEventType.WATCHED)) {
                      WatchedEvent event = e.getWatchedEvent();
  
                      watcher.execute(event.getState(), event.getType(), event.getPath());
                  }
  
              }
          });
  
          fk.getUnhandledErrorListenable().addListener(new UnhandledErrorListener() {
              @Override
              public void unhandledError(String msg, Throwable error) {
                  String errmsg = "Unrecoverable zookeeper error, halting process: " + msg;
                  LOG.error(errmsg, error);
                  JStormUtils.halt_process(1, "Unrecoverable zookeeper error");
  
              }
          });
          fk.start();
          return fk;
      }
docs: connect ZK, register watchers