docs: def ready(self, cluster):
          """
          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()
  
          exhausted = bool(self._free.queued() > 0)
          # 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_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:
                      # 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