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); int part = create(_admin, _zkClient, _clusterName, _numPartitions); 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"); } _watcher = new DatabusHelixWatcher(); } DatabusCluster(ClusterRegistrationStaticConfig config); protected  DatabusCluster(); static int create(ZKHelixAdmin admin, ZkClient zkClient, String clusterName,
            int numPartitions); void start(); void shutdown(); DatabusClusterMember addMember(String id); DatabusClusterMember addMember(String id,
            DatabusClusterNotifier notifier); synchronized void addDataNotifier(DatabusClusterDataNotifier notifier); synchronized void removeDataNotifier(
            DatabusClusterDataNotifier notifier); int getNumPartitions(); int getNumActiveMembers(); int getNumActivePartitions(); HashMap<Integer, String> getActivePartitions(); String getPartitionOwner(int partition); String getClusterName(); int getQuorum(); @Override String toString(); static final String MODULE; static final Logger LOG; static final String DEFAULT_STATE_MODEL; static final String DEFAULT_RESOURCE_NAME; static final int DEFAULT_CLUSTER_CREATE_WAIT_MS; }
@Test public void testDatabusCluster() { runBasicDatabusClusterExpt(false); }