code: def _setup_kafka(self):
          
          # close older connections
          if self.consumer is not None:
              self.logger.debug("Closing existing kafka consumer")
              self.consumer.close()
              self.consumer = None
          if self.producer is not None:
              self.logger.debug("Closing existing kafka producer")
              self.producer.flush()
              self.producer.close(timeout=10)
              self.producer = None
  
          # create new connections
          self._consumer_thread = None
          self.logger.debug("Creating kafka connections")
          self.consumer = self._create_consumer()
          if not self.closed:
              self.logger.debug("Kafka Conumer created")
          self.producer = self._create_producer()
          if not self.closed:
              self.logger.debug("Kafka Producer created")
  
          if not self.closed:
              self.kafka_connected = True
              self.logger.info("Connected successfully to Kafka")
              self._spawn_kafka_consumer_thread()
docs: Sets up kafka connections