class MasterRegistry {

    public void registry() {
        String address = NetUtils.getHost();
        String localNodePath = getMasterPath();
        zookeeperRegistryCenter.getZookeeperCachedOperator().persistEphemeral(localNodePath, "");
        zookeeperRegistryCenter.getZookeeperCachedOperator().getZkClient().getConnectionStateListenable().addListener(new ConnectionStateListener() {
            @Override
            public void stateChanged(CuratorFramework client, ConnectionState newState) {
                if (newState == ConnectionState.LOST) {
                    logger.error("master : {} connection lost from zookeeper", address);
                } else if (newState == ConnectionState.RECONNECTED) {
                    logger.info("master : {} reconnected to zookeeper", address);
                    zookeeperRegistryCenter.getZookeeperCachedOperator().persistEphemeral(localNodePath, "");
                } else if (newState == ConnectionState.SUSPENDED) {
                    logger.warn("master : {} connection SUSPENDED ", address);
                }
            }
        });
        int masterHeartbeatInterval = masterConfig.getMasterHeartbeatInterval();
        HeartBeatTask heartBeatTask = new HeartBeatTask(startTime,
                masterConfig.getMasterReservedMemory(),
                masterConfig.getMasterMaxCpuloadAvg(),
                Sets.newHashSet(getMasterPath()),
                zookeeperRegistryCenter);

        this.heartBeatExecutor.scheduleAtFixedRate(heartBeatTask, 0, masterHeartbeatInterval, TimeUnit.SECONDS);
        logger.info("master node : {} registry to ZK path {} successfully with heartBeatInterval : {}s"
                , address, localNodePath, masterHeartbeatInterval);
    }

}