class_docstr: Accepts messages from Kafka
  
  Uses the confluent-kafka library,
  https://docs.confluent.io/current/clients/confluent-kafka-python/
  
  
  Parameters
  ----------
  topics: list of str
      Labels of Kafka topics to consume from
  consumer_params: dict
      Settings to set up the stream, see
      https://docs.confluent.io/current/clients/confluent-kafka-python/#configuration
      https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
      Examples:
      bootstrap.servers, Connection string(s) (host:port) by which to reach
      Kafka;
      group.id, Identity of the consumer. If multiple sources share the same
      group, each message will be passed to only one of them.
  poll_interval: number
      Seconds that elapse between polling Kafka for new messages
  start: bool (False)
      Whether to start polling upon instantiation
  
  Examples
  --------
  
  >>> source = Stream.from_kafka(['mytopic'],
  ...           {'bootstrap.servers': 'localhost:9092',
  ...            'group.id': 'streamz'})  # doctest: +SKIP
class_docstr_tok: ["Accepts","messages","from","Kafka","Uses","the","confluent-kafka","library",",","https",":","\/\/docs.confluent.io\/current\/clients\/confluent-kafka-python\/","Parameters","--","--","--","--","--","topics",":","list","of","str","Labels","of","Kafka","topics","to","consume","from","consumer_params",":","dict","Settings","to","set","up","the","stream",",","see","https",":","\/\/docs.confluent.io\/current\/clients\/confluent-kafka-python\/","#","configuration","https",":","\/\/github.com\/edenhill\/librdkafka\/blob\/master\/CONFIGURATION.md","Examples",":","bootstrap.servers",",","Connection","string","(","s",")","(","host",":","port",")","by","which","to","reach","Kafka",";","group.id",",","Identity","of","the","consumer",".","If","multiple","sources","share","the","same","group",",","each","message","will","be","passed","to","only","one","of","them",".","poll_interval",":","number","Seconds","that","elapse","between","polling","Kafka","for","new","messages","start",":","bool","(","False",")","Whether","to","start","polling","upon","instantiation","Examples","--","--","--","--",">",">",">","source","=","Stream.from_kafka","(","[","'mytopic","'","]",",","...","{","'bootstrap.servers","'",":","'localhost:9092","'",",","...","'group.id","'",":","'streamz","'","}",")","#","doctest",":","+SKIP"]
code: def poll_kafka(self):
      while True:
          val = self.do_poll()
          if val:
              yield self._emit(val)
          else:
              yield gen.sleep(self.poll_interval)
          if self.stopped:
              break
      self._close_consumer()
  
imports: import asyncio
  from glob import glob
  import queue
  import os
  import time
  from tornado import gen
  import weakref
  from .core import Stream, convert_interval, RefCounter, sync
label_desc: Use image node_id 3 for calling the from_kafka obj's underlying member method code with example usage: obj.poll_kafka() without return types