code: def makeRequest(self, requestId, request, expectResponse=True):
          
          if requestId in self.requests:
              # Id is duplicate to 'in-flight' request. Reject it, as we
              # won't be able to properly deliver the response(s)
              # Note that this won't protect against a client calling us
              # twice with the same ID, but first with expectResponse=False
              # But that's pathological, and the only defense is to track
              # all requestIds sent regardless of whether we expect to see
              # a response, which is effectively a memory leak...
              raise DuplicateRequestError(
                  'Reuse of requestId:{}'.format(requestId))
  
          # If we've been told to shutdown (close() called) then fail request
          if self._dDown:
              return fail(ClientError('makeRequest() called after close()'))
  
          # Ok, we are going to save/send it, create a _Request object to track
          canceller = partial(
              self.cancelRequest, requestId,
              CancelledError(message="Request correlationId={} was cancelled".format(requestId)))
          tReq = _Request(requestId, request, expectResponse, canceller)
  
          # add it to our requests dict
          self.requests[requestId] = tReq
  
          # Add an errback to the tReq.d to remove it from our requests dict
          # if something goes wrong...
          tReq.d.addErrback(self._handleRequestFailure, requestId)
  
          # Do we have a connection over which to send the request?
          if self.proto:
              # Send the request
              self._sendRequest(tReq)
          # Have we not even started trying to connect yet? Do so now
          elif not self.connector:
              self._connect()
          return tReq.d
docs: Send a request to our broker via our self.proto KafkaProtocol object.
  
          Return a deferred which will fire when the reply matching the requestId
          comes back from the server, or, if expectResponse is False, then
          return None instead.
          If we are not currently connected, then we buffer the request to send
          when the connection comes back up.