code: def abort_expired_batches(self, request_timeout_ms, cluster):
          
          expired_batches = []
          to_remove = []
          count = 0
          for tp in list(self._batches.keys()):
              assert tp in self._tp_locks, 'TopicPartition not in locks dict'
  
              # We only check if the batch should be expired if the partition
              # does not have a batch in flight. This is to avoid the later
              # batches get expired when an earlier batch is still in progress.
              # This protection only takes effect when user sets
              # max.in.flight.request.per.connection=1. Otherwise the expiration
              # order is not guranteed.
              if tp in self.muted:
                  continue
  
              with self._tp_locks[tp]:
                  # iterate over the batches and expire them if they have stayed
                  # in accumulator for more than request_timeout_ms
                  dq = self._batches[tp]
                  for batch in dq:
                      is_full = bool(bool(batch != dq[-1]) or batch.records.is_full())
                      # check if the batch is expired
                      if batch.maybe_expire(request_timeout_ms,
                                            self.config['retry_backoff_ms'],
                                            self.config['linger_ms'],
                                            is_full):
                          expired_batches.append(batch)
                          to_remove.append(batch)
                          count += 1
                          self.deallocate(batch)
                      else:
                          # Stop at the first batch that has not expired.
                          break
  
                  # Python does not allow us to mutate the dq during iteration
                  # Assuming expired batches are infrequent, this is better than
                  # creating a new copy of the deque for iteration on every loop
                  if to_remove:
                      for batch in to_remove:
                          dq.remove(batch)
                      to_remove = []
  
          if expired_batches:
              log.warning("Expired %d batches in accumulator", count) # trace
  
          return expired_batches
docs: Abort the batches that have been sitting in RecordAccumulator for
          more than the configured request_timeout due to metadata being
          unavailable.
  
          Arguments:
              request_timeout_ms (int): milliseconds to timeout
              cluster (ClusterMetadata): current metadata for kafka cluster
  
          Returns:
              list of ProducerBatch that were expired