Force auth when fetching Kafka metadata

Author: andrei-ionescu

src/consumer/base_consumer.rs

@@ -713,7 +713,10 @@ where
         topic: Option<&str>,
         timeout: T,
     ) -> KafkaResult<Metadata> {
-        self.client.fetch_metadata(topic, timeout)
+        let t = timeout.into();
+        // force credential retrieval
+        self.client.poll_event(&self.queue, t);
+        self.client.fetch_metadata(topic, t)
     }
 
     fn fetch_watermarks<T: Into<Timeout>>(