diff --git a/ceilometer/collector/service.py b/ceilometer/collector/service.py
index abe011d50..51a375226 100644
--- a/ceilometer/collector/service.py
+++ b/ceilometer/collector/service.py
@@ -60,6 +60,8 @@ class CollectorService(service.PeriodicService):
         self.storage_engine = storage.get_engine(cfg.CONF)
         self.storage_conn = self.storage_engine.get_connection(cfg.CONF)
 
+    def initialize_service_hook(self, service):
+        '''Consumers must be declared before consume_thread start.'''
         self.ext_manager = extension_manager.ActivatedExtensionManager(
             namespace=self.COLLECTOR_NAMESPACE,
             disabled_names=cfg.CONF.disabled_notification_listeners,
diff --git a/ceilometer/openstack/common/rpc/service.py b/ceilometer/openstack/common/rpc/service.py
index ab7d630fd..6d653434f 100644
--- a/ceilometer/openstack/common/rpc/service.py
+++ b/ceilometer/openstack/common/rpc/service.py
@@ -57,6 +57,11 @@ class Service(service.Service):
 
         self.conn.create_consumer(self.topic, dispatcher, fanout=True)
 
+        # Hook to allow the manager to do other initializations after
+        # the rpc connection is created.
+        if callable(getattr(self.manager, 'initialize_service_hook', None)):
+            self.manager.initialize_service_hook(self)
+
         # Consume from all consumers in a thread
         self.conn.consume_in_thread()