class_docstr: The background thread that handles the sending of produce requests to the
  Kafka cluster. This thread makes metadata requests to renew its view of the
  cluster and then sends produce requests to the appropriate nodes.
docstr: Handle a produce response.
code: def _handle_produce_response(
      self, node_id, send_time, batches, response
  ):
      """Handle a produce response."""
      # if we have a response, parse it
      log.debug("Parsing produce response: %r", response)
      if response:
          batches_by_partition = dict(
              [(batch.topic_partition, batch) for batch in batches]
          )
  
          for topic, partitions in response.topics:
              for partition_info in partitions:
                  global_error = None
                  log_start_offset = None
                  if response.API_VERSION < 2:
                      partition, error_code, offset = partition_info
                      ts = None
                  elif 2 <= response.API_VERSION <= 4:
                      partition, error_code, offset, ts = partition_info
                  elif 5 <= response.API_VERSION <= 7:
                      (
                          partition,
                          error_code,
                          offset,
                          ts,
                          log_start_offset,
                      ) = partition_info
                  else:
                      # the ignored parameter is record_error of type list[(batch_index: int, error_message: str)]
                      (
                          partition,
                          error_code,
                          offset,
                          ts,
                          log_start_offset,
                          _,
                          global_error,
                      ) = partition_info
                  tp = TopicPartition(topic, partition)
                  error = Errors.for_code(error_code)
                  batch = batches_by_partition[tp]
                  self._complete_batch(
                      batch,
                      error,
                      offset,
                      ts,
                      log_start_offset,
                      global_error,
                  )
  
          if response.API_VERSION > 0:
              self._sensors.record_throttle_time(
                  response.throttle_time_ms, node=node_id
              )
  
      else:
          # this is the acks = 0 case, just complete all requests
          for batch in batches:
              self._complete_batch(batch, None, -1, None)
  
imports: from __future__ import absolute_import, division
  import collections
  import copy
  import logging
  import threading
  import time
  from kafka.vendor import six
  from kafka import errors
  from kafka.metrics.measurable import AnonMeasurable
  from kafka.metrics.stats import Avg, Max, Rate
  from kafka.protocol.produce import ProduceRequest
  from kafka.structs import TopicPartition
  from kafka.version import __version__
label_desc: Use image node_id 8 for calling the Sender obj's underlying member method code with example usage: obj._handle_produce_response(node_id, send_time, batches, response) without return types