_log.error("ConsumerFactory for cluster (" + _clusterInfo + ") returned null or empty consumers ");
throw new DatabusClientException("ConsumerFactory for cluster (" + _clusterInfo + ") returned null or empty consumers");
}
// Create Registration
RegistrationId id = new RegistrationId(_id + "-" + partition.getPartitionId());
CheckpointPersistenceProvider ckptProvider = createCheckpointPersistenceProvider(partition);
DatabusV2RegistrationImpl reg = createChildRegistration(id, _client, ckptProvider);
reg.addDatabusConsumers(consumers);
String[] srcs = new String[_sources.size()];
srcs = _sources.toArray(srcs);
reg.addSubscriptions(srcs);
regMap.put(partition,reg);
reg.onRegister();
// Add Server-Side Filter
if ( null != filterConfig)
reg.withServerSideFilter(filterConfig);
// Notify Listener
if (null != _partitionListener)
_partitionListener.onAddPartition(partition, reg);
// Start the registration
try {
reg.start();
} catch (DatabusClientException e) {
_log.error("Got exception while starting the registration for partition (" + partition + ")", e);
throw e;
}
//Add partition Specific metrics to cluster-merge
_relayEventStatsMerger.addStatsCollector(id.getId(), (DbusEventsStatisticsCollector)reg.getRelayEventStats());
_bootstrapEventStatsMerger.addStatsCollector(id.getId(), (DbusEventsStatisticsCollector)reg.getBootstrapEventStats());
_relayCallbackStatsMerger.addStatsCollector(id.getId(), (ConsumerCallbackStats)reg.getRelayCallbackStats());
_bootstrapCallbackStatsMerger.addStatsCollector(id.getId(), (ConsumerCallbackStats)reg.getBootstrapCallbackStats());
_log.info("Partition (" + partition + ") started !!");
} catch (DatabusException e) {
_log.error("Got exception while activating partition(" + partition + ")",e);
throw new DatabusClientException(e);