'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'

def ready(self, cluster):

    ready_nodes = set()
    next_ready_check = 9999999.99
    unknown_leaders_exist = False
    now = time.time()
    exhausted = bool((self._free.queued() > 0))
    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_locks[tp]:
            dq = self._batches[tp]
            if (not dq):
                continue
            batch = dq[0]
            retry_backoff = (self.config['retry_backoff_ms'] / 1000.0)
            linger = (self.config['linger_ms'] / 1000.0)
            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 exhausted or self._closed or self._flush_in_progress())
            if (sendable and (not backing_off)):
                ready_nodes.add(leader)
            else:
                next_ready_check = min(time_left, next_ready_check)
    return (ready_nodes, next_ready_check, unknown_leaders_exist)
