code: def _cancel_send_messages(self, d):
          
          # Is the request in question in an unsent batch?
          for req in self._batch_reqs:
              if req.deferred == d:
                  # Found the request, remove it and return.
                  msgs = req.messages
                  self._waitingMsgCount -= len(msgs)
                  for m in (_m for _m in msgs if _m is not None):
                      self._waitingByteCount -= len(m)
                  # This _should_ be safe as we abort the iteration upon removal
                  self._batch_reqs.remove(req)
                  d.errback(CancelledError(request_sent=False))
                  return
  
          # If it wasn't found in the unsent batch. We just rely on the
          # downstream processing of the request to check if the deferred
          # has been called and skip further processing for this request
          # Errback the deferred with whether or not we sent the request
          # to Kafka already
          d.errback(
              CancelledError(request_sent=(self._batch_send_d is not None)))
          return
docs: Cancel a `send_messages` request
          First check if the request is in a waiting batch, of so, great, remove
          it from the batch. If it's not found, we errback() the deferred and
          the downstream processing steps take care of aborting further
          processing.
          We check if there's a current _batch_send_d to determine where in the
          chain we were (getting partitions, or already sent request to Kafka)
          and errback differently.