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: Drain all the data for the given nodes and collate them into a list of
  batches that will fit within the specified size on a per-node basis.
  This method attempts to avoid choosing the same topic-node repeatedly.
  
  Arguments:
      cluster (ClusterMetadata): The current cluster metadata
      nodes (list): list of node_ids to drain
      max_size (int): maximum number of bytes to drain
  
  Returns:
      dict: {node_id: list of ProducerBatch} with total size less than the
          requested max_size.
docstr_tok: ["Drain","all","the","data","for","the","given","nodes","and","collate","them","into","a","list","of","batches","that","will","fit","within","the","specified","size","on","a","per-node","basis",".","This","method","attempts","to","avoid","choosing","the","same","topic-node","repeatedly",".","Arguments",":","cluster","(","ClusterMetadata",")",":","The","current","cluster","metadata","nodes","(","list",")",":","list","of","node_ids","to","drain","max_size","(","int",")",":","maximum","number","of","bytes","to","drain","Returns",":","dict",":","{","node_id",":","list","of","ProducerBatch","}","with","total","size","less","than","the","requested","max_size","."]
code: def drain(self, cluster, nodes, max_size):
      """
      Drain all the data for the given nodes and collate them into a list of
      batches that will fit within the specified size on a per-node basis.
      This method attempts to avoid choosing the same topic-node repeatedly.
  
      Arguments:
          cluster (ClusterMetadata): The current cluster metadata
          nodes (list): list of node_ids to drain
          max_size (int): maximum number of bytes to drain
  
      Returns:
          dict: {node_id: list of ProducerBatch} with total size less than the
              requested max_size.
      """
      if not nodes:
          return {}
  
      now = time.time()
      batches = {}
      for node_id in nodes:
          size = 0
          partitions = list(cluster.partitions_for_broker(node_id))
          ready = []
          # to make starvation less likely this loop doesn't start at 0
          self._drain_index %= len(partitions)
          start = self._drain_index
          while True:
              tp = partitions[self._drain_index]
              if tp in self._batches and tp not in self.muted:
                  with self._tp_locks[tp]:
                      dq = self._batches[tp]
                      if dq:
                          first = dq[0]
                          backoff = bool(first.attempts > 0) and bool(
                              first.last_attempt
                              + self.config["retry_backoff_ms"] / 1000.0
                              > now
                          )
                          # Only drain the batch if it is not during backoff
                          if not backoff:
                              if (
                                  size + first.records.size_in_bytes()
                                  > max_size
                                  and len(ready) > 0
                              ):
                                  # there is a rare case that a single batch
                                  # size is larger than the request size due
                                  # to compression; in this case we will
                                  # still eventually send this batch in a
                                  # single request
                                  break
                              else:
                                  batch = dq.popleft()
                                  batch.records.close()
                                  size += batch.records.size_in_bytes()
                                  ready.append(batch)
                                  batch.drained = now
  
              self._drain_index += 1
              self._drain_index %= len(partitions)
              if start == self._drain_index:
                  break
  
          batches[node_id] = ready
      return batches
  
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 7 for calling the RecordAccumulator obj's underlying member method code with example usage: obj.drain(cluster, nodes, max_size) and returns: batches, dict