def ready(self, cluster, now=None):
        """
        Get a list of nodes whose partitions are ready to be sent, and the
        earliest time at which any non-sendable partition will be ready;
        Also return the flag for whether there are any unknown leaders for the
        accumulated partition batches.

        A destination node is ready to send if:

         * There is at least one partition that is not backing off its send
         * and those partitions are not muted (to prevent reordering if
           max_in_flight_requests_per_connection is set to 1)
         * and any of the following are true:

           * The record set is full
           * The record set has sat in the accumulator for at least linger_ms
             milliseconds
           * The accumulator is out of memory and threads are blocking waiting
             for data (in this case all partitions are immediately considered
             ready).
           * The accumulator has been closed

        Arguments:
            cluster (ClusterMetadata):

        Returns:
            tuple:
                ready_nodes (set): node_ids that have ready batches
                next_ready_check (float): secs until next ready after backoff
                unknown_leaders_exist (bool): True if metadata refresh needed
        """
        ready_nodes = set()
        next_ready_check = 9999999.99
        unknown_leaders_exist = False
        now = time.time() if now is None else now

        # several threads are accessing self._batches -- to simplify
        # concurrent access, we iterate over a snapshot of partitions
        # and lock each partition separately as needed
        partitions = list(self._batches.keys())
        for tp in partitions:
            leader = cluster.leader_for_partition(tp)
            if leader is None or leader == -1:
                unknown_leaders_exist = True
                continue
            elif leader in ready_nodes:
                continue
            elif tp in self.muted:
                continue

            with self._tp_lock(tp):
                dq = self._batches[tp]
                if not dq:
                    continue
                batch = dq[0]
                retry_backoff = self.config['retry_backoff_ms'] / 1000
                linger = self.config['linger_ms'] / 1000
                backing_off = bool(batch.attempts > 0
                                   and (batch.last_attempt + retry_backoff) > now)
                waited_time = now - batch.last_attempt
                time_to_wait = retry_backoff if backing_off else linger
                time_left = max(time_to_wait - waited_time, 0)
                full = bool(len(dq) > 1 or batch.records.is_full())
                expired = bool(waited_time >= time_to_wait)

                sendable = (full or expired or self._closed or
                            self._flush_in_progress())

                if sendable and not backing_off:
                    ready_nodes.add(leader)
                else:
                    # Note that this results in a conservative estimate since
                    # an un-sendable partition may have a leader that will
                    # later be found to have sendable data. However, this is
                    # good enough since we'll just wake up and then sleep again
                    # for the remaining time.
                    next_ready_check = min(time_left, next_ready_check)

        return ready_nodes, next_ready_check, unknown_leaders_exist

        Get a list of nodes whose partitions are ready to be sent, and the
        earliest time at which any non-sendable partition will be ready;
        Also return the flag for whether there are any unknown leaders for the
        accumulated partition batches.

        A destination node is ready to send if:

         * There is at least one partition that is not backing off its send
         * and those partitions are not muted (to prevent reordering if
           max_in_flight_requests_per_connection is set to 1)
         * and any of the following are true:

           * The record set is full
           * The record set has sat in the accumulator for at least linger_ms
             milliseconds
           * The accumulator is out of memory and threads are blocking waiting
             for data (in this case all partitions are immediately considered
             ready).
           * The accumulator has been closed

        Arguments:
            cluster (ClusterMetadata):

        Returns:
            tuple:
                ready_nodes (set): node_ids that have ready batches
                next_ready_check (float): secs until next ready after backoff
                unknown_leaders_exist (bool): True if metadata refresh needed
        
kafka/producer/record_accumulator.py