class DatabusCluster {

    public DatabusCluster(ClusterRegistrationStaticConfig config) throws Exception
    {
        _zkAddr = config.getZkAddr();
        _clusterName = config.getClusterName();
        _quorum = (int)(config.getQuorum());
        _numPartitions = (int)(config.getNumPartitions());

        _zkSessionTimeoutMs = config.getZkSessionTimeoutMs();
        _zkConnectionTimeoutMs = config.getZkConnectionTimeoutMs();

        updateHelixManagerZkSessionTimeout(_zkSessionTimeoutMs);

        _zkClient = new ZkClient(_zkAddr, _zkSessionTimeoutMs, _zkConnectionTimeoutMs, new ZNRecordSerializer());
        _admin = new ZKHelixAdmin(_zkClient);
        _dataNotifiers = new HashSet<DatabusClusterDataNotifier>(5);

        //attempt to create a cluster
        int part = create(_admin, _zkClient, _clusterName, _numPartitions);

        //at this stage a cluster and resources should have been created, either by this instance or someone else
        if (part >= 0)
        {
            if (_numPartitions != part)
            {
                String msg = "Cannot create DatabusCluster!  Cluster exists with num partitions="
                        + part
                        + ". Tried to join with "
                        + _numPartitions
                        + " partitions";
                throw new DatabusClusterException(msg);
            }
        }
        else
        {
            throw new DatabusClusterException("Cluster " + _clusterName + " could not be accessed. Num partitions returned -1");
        }

        // initialize watcher after creating a cluster
        _watcher = new DatabusHelixWatcher();
    }

    public  DatabusCluster(ClusterRegistrationStaticConfig config);
    protected  DatabusCluster();

}