subscribe to input topic

src/bin/processor.rs

@@ -155,6 +155,10 @@ fn main() -> Result<()> {
     )
     .context("producer built")?;
 
+    consumer
+        .subscribe(&[input_topic])
+        .expect("topic subscription failed");
+
     let mut counter = Counter::new();
     loop {
         for msg_result in consumer.iter() {