'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 RecordBatch that were expired'

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'
        if (tp in self.muted):
            continue
        with self._tp_locks[tp]:
            dq = self._batches[tp]
            for batch in dq:
                is_full = bool((bool((batch != dq[(-1)])) or batch.records.is_full()))
                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:
                    break
            if to_remove:
                for batch in to_remove:
                    dq.remove(batch)
                to_remove = []
    if expired_batches:
        log.debug('Expired    %d    batches    in    accumulator', count)
    return expired_batches
