code: def _check_kafka_disconnect(self):
          
          for node_id in self.consumer._client._conns:
              conn = self.consumer._client._conns[node_id]
              if conn.state == ConnectionStates.DISCONNECTED or \
                      conn.state == ConnectionStates.DISCONNECTING:
                  self._spawn_kafka_connection_thread()
                  break
docs: Checks the kafka connection is still valid