Add list_topics call to custreamz Kafka reader

Author: chinmaychandak

streamz/sources.py

@@ -486,7 +486,10 @@ def checkpoint_emit(_part):
 
         if self.npartitions is None:
             kafka_cluster_metadata = self.consumer.list_topics(self.topic)
-            self.npartitions = len(kafka_cluster_metadata.topics[self.topic].partitions)
+            if self.engine == "cudf":  # pragma: no cover
+                self.npartitions = len(kafka_cluster_metadata[self.topic.encode('utf-8')])
+            else:
+                self.npartitions = len(kafka_cluster_metadata.topics[self.topic].partitions)
         self.positions = [0] * self.npartitions
 
         tps = []