code: public void registerBrokerInZk() {
          logger.info("Registering broker " + brokerIdPath);
          String hostname = config.getHostName();
          if (hostname == null) {
              try {
                  hostname = InetAddress.getLocalHost().getHostName();
              } catch (UnknownHostException e) {
                  throw new RuntimeException("cannot get local host, setting 'hostname' in configuration");
              }
          }
          //
          final String creatorId = hostname + "-" + System.currentTimeMillis();
          final Broker broker = new Broker(config.getBrokerId(), creatorId, hostname, config.getPort(),config.isTopicAutoCreated());
          try {
              ZkUtils.createEphemeralPathExpectConflict(zkClient, brokerIdPath, broker.getZKString());
          } catch (ZkNodeExistsException e) {
              String oldServerInfo = ZkUtils.readDataMaybeNull(zkClient, brokerIdPath);
              String message = "A broker (%s) is already registered on the path %s." //
                      + " This probably indicates that you either have configured a brokerid that is already in use, or "//
                      + "else you have shutdown this broker and restarted it faster than the zookeeper " ///
                      + "timeout so it appears to be re-registering.";
              message = String.format(message, oldServerInfo, brokerIdPath);
              throw new RuntimeException(message);
          }
          //
          logger.info("Registering broker " + brokerIdPath + " succeeded with " + broker);
      }
docs: register broker in the zookeeper
  <p>
  path: /brokers/ids/&lt;id&gt; <br>
  data: creator:host:port