kafka producer send stats callback ensure message delivery

kafka_send_statistics.py

stats = {
      "errors": 0,
      "successes": 0,
      "min_offset": None,
      "max_offset": None,
      "errorTypes": {}
}

def errback_listener(error, **kargs):
      if type(error).__name__ not in kargs["stats"]["errorTypes"]:
          kargs["stats"]["errorTypes"][type(error).__name__] = 0
      kargs["stats"]["errorTypes"][type(error).__name__] += 1
      kargs["stats"]["errors"] += 1

future = producer.send(task["kafka"]["topic"], key=bytes('no_partition', encoding='utf-8'), value=bytes(json.dumps(record), encoding='utf-8'))
future.add_errback(errback_listener,stats=stats)