def _get_expired_inflight_batches(self, now=None):
        """Get the in-flight batches that has reached delivery timeout."""
        expired_batches = []
        to_remove = []
        for tp, queue in six.iteritems(self._in_flight_batches):
            while queue:
                _created_at, batch = queue[0]
                if batch.has_reached_delivery_timeout(self._accumulator.delivery_timeout_ms):
                    heapq.heappop(queue)
                    if batch.final_state is None:
                        expired_batches.append(batch)
                    else:
                        raise Errors.IllegalStateError("%s batch created at %s gets unexpected final state %s" % (batch.topic_partition, batch.created, batch.final_state))
                else:
                    self._accumulator.maybe_update_next_batch_expiry_time(batch)
                    break
            else:
                # Avoid mutating in_flight_batches during iteration
                to_remove.append(tp)
        for tp in to_remove:
            del self._in_flight_batches[tp]
        return expired_batches
Get the in-flight batches that has reached delivery timeout.
kafka/producer/sender.py