from kafka import KafkaConsumer
class KafkaMonitor(object):
    def __init__(self):
        self.consumer = KafkaConsumer("btest", bootstrap_servers="172.20.68.142:9093")

    def run(self):
        for message in self.consumer:
            print(str(message))
if __name__ == "__main__":
    KafkaMonitor().run()

def poll(self, timeout_ms=None, future=None):
...
  # check the ip\port in conn are consistent with ip\port in metadata
  for node_id, conn in self._conns.items():
      broker = self.cluster.broker_metadata(conn.node_id)
      if broker is None:
          self.close(node_id)
          continue
  
      host, _, afi = get_ip_port_afi(broker.host)
      if conn.host != host or conn.port != broker.port:
          self.close(node_id)
          continue

def _register_send_sockets(self):
    while self._sending:
        conn = self._sending.pop()
        if conn.disconnected():
            continue
        try:
            key = self._selector.get_key(conn._sock)
            events = key.events | selectors.EVENT_WRITE
            self._selector.modify(key.fileobj, events, key.data)
        except KeyError:
            self._selector.register(conn._sock, selectors.EVENT_WRITE, conn)