diff --git a/trustgraph/base/consumer_producer.py b/trustgraph/base/consumer_producer.py
index 08ae1880..cabb7525 100644
--- a/trustgraph/base/consumer_producer.py
+++ b/trustgraph/base/consumer_producer.py
@@ -63,16 +63,16 @@ class ConsumerProducer(BaseProcessor):
         if output_schema == None:
             raise RuntimeError("output_schema must be specified")
 
-        self.consumer = self.client.subscribe(
-            input_queue, subscriber,
-            schema=JsonSchema(input_schema),
-        )
-
         self.producer = self.client.create_producer(
             topic=output_queue,
             schema=JsonSchema(output_schema),
         )
 
+        self.consumer = self.client.subscribe(
+            input_queue, subscriber,
+            schema=JsonSchema(input_schema),
+        )
+
     def run(self):
 
         __class__.state_metric.state('running')