<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 def abort_expired_batches(self, request_timeout_ms, cluster):
 """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 """

        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