def expired_batches(self, now=None):
        """Get a list of batches which have been sitting in the accumulator too long and need to be expired."""
        expired_batches = []
        for tp in list(self._batches.keys()):
            with self._tp_lock(tp):
                # iterate over the batches and expire them if they have stayed
                # in accumulator for more than request_timeout_ms
                dq = self._batches[tp]
                while dq:
                    batch = dq[0]
                    if batch.has_reached_delivery_timeout(self.delivery_timeout_ms, now=now):
                        dq.popleft()
                        batch.records.close()
                        expired_batches.append(batch)
                    else:
                        # Stop at the first batch that has not expired.
                        self.maybe_update_next_batch_expiry_time(batch)
                        break
        return expired_batches
Get a list of batches which have been sitting in the accumulator too long and need to be expired.
kafka/producer/record_accumulator.py