class_docstr: This class maintains a dequeue per TopicPartition that accumulates messages
  into MessageSets to be sent to the server.
  
  The accumulator attempts to bound memory use, and append calls will block
  when that memory is exhausted.
  
  Keyword Arguments:
      batch_size (int): Requests sent to brokers will contain multiple
          batches, one for each partition with data available to be sent.
          A small batch size will make batching less common and may reduce
          throughput (a batch size of zero will disable batching entirely).
          Default: 16384
      buffer_memory (int): The total bytes of memory the producer should use
          to buffer records waiting to be sent to the server. If records are
          sent faster than they can be delivered to the server the producer
          will block up to max_block_ms, raising an exception on timeout.
          In the current implementation, this setting is an approximation.
          Default: 33554432 (32MB)
      compression_attrs (int): The compression type for all data generated by
          the producer. Valid values are gzip(1), snappy(2), lz4(3), or
          none(0).
          Compression is of full batches of data, so the efficacy of batching
          will also impact the compression ratio (more batching means better
          compression). Default: None.
      linger_ms (int): An artificial delay time to add before declaring a
          messageset (that isn't full) ready for sending. This allows
          time for more records to arrive. Setting a non-zero linger_ms
          will trade off some latency for potentially better throughput
          due to more batching (and hence fewer, larger requests).
          Default: 0
      retry_backoff_ms (int): An artificial delay time to retry the
          produce request upon receiving an error. This avoids exhausting
          all retries in a short period of time. Default: 100
class_docstr_tok: ["This","class","maintains","a","dequeue","per","TopicPartition","that","accumulates","messages","into","MessageSets","to","be","sent","to","the","server",".","The","accumulator","attempts","to","bound","memory","use",",","and","append","calls","will","block","when","that","memory","is","exhausted",".","Keyword","Arguments",":","batch_size","(","int",")",":","Requests","sent","to","brokers","will","contain","multiple","batches",",","one","for","each","partition","with","data","available","to","be","sent",".","A","small","batch","size","will","make","batching","less","common","and","may","reduce","throughput","(","a","batch","size","of","zero","will","disable","batching","entirely",")",".","Default",":","16384","buffer_memory","(","int",")",":","The","total","bytes","of","memory","the","producer","should","use","to","buffer","records","waiting","to","be","sent","to","the","server",".","If","records","are","sent","faster","than","they","can","be","delivered","to","the","server","the","producer","will","block","up","to","max_block_ms",",","raising","an","exception","on","timeout",".","In","the","current","implementation",",","this","setting","is","an","approximation",".","Default",":","33554432","(","32MB",")","compression_attrs","(","int",")",":","The","compression","type","for","all","data","generated","by","the","producer",".","Valid","values","are","gzip","(","1",")",",","snappy","(","2",")",",","lz4","(","3",")",",","or","none","(","0",")",".","Compression","is","of","full","batches","of","data",",","so","the","efficacy","of","batching","will","also","impact","the","compression","ratio","(","more","batching","means","better","compression",")",".","Default",":","None",".","linger_ms","(","int",")",":","An","artificial","delay","time","to","add","before","declaring","a","messageset","(","that","is","n't","full",")","ready","for","sending",".","This","allows","time","for","more","records","to","arrive",".","Setting","a","non-zero","linger_ms","will","trade","off","some","latency","for","potentially","better","throughput","due","to","more","batching","(","and","hence","fewer",",","larger","requests",")",".","Default",":","0","retry_backoff_ms","(","int",")",":","An","artificial","delay","time","to","retry","the","produce","request","upon","receiving","an","error",".","This","avoids","exhausting","all","retries","in","a","short","period","of","time",".","Default",":","100"]
docstr: 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
docstr_tok: ["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"]
code: 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
  
imports: from __future__ import absolute_import
  import collections
  import copy
  import logging
  import threading
  import time
  import kafka.errors
  from kafka.producer.buffer import SimpleBufferPool
  from kafka.producer.future import FutureRecordMetadata, FutureProduceResult
  from kafka.record.memory_records import MemoryRecordsBuilder
  from kafka.structs import TopicPartition
label_desc: Use image node_id 5 for calling the RecordAccumulator obj's underlying member method code with example usage: obj.ready(cluster) and returns: ready_nodes,  next_ready_check,  unknown_leaders_exist