code: def request(self, request):
          
          if self._failed is not None:
              return fail(self._failed)
          correlation_id = request[4:8]
          assert correlation_id not in self._pending
          d = Deferred()
          self.sendString(request)
          self._pending[correlation_id] = d
          return d
docs: Send a request to the Kafka broker.
  
          :param bytes request:
              The bytes of a Kafka `RequestMessage`_ structure. It must have
              a unique (to this connection) correlation ID.
  
          :returns:
              `Deferred` which will:
  
                - Succeed with the bytes of a Kafka `ResponseMessage`_
                - Fail when the connection terminates
  
          .. _RequestMessage:: https://kafka.apache.org/protocol.html#protocol_messages