producer

var GenDefaultDeliveryHandlerFuncWithZapLogger = func(log *zap.Logger) func(event kafka.Event) {
	return func(e kafka.Event) {
		switch ev := e.(type) {
		case *kafka.Message:
			if ev.TopicPartition.Error != nil {
				log.Error(fmt.Sprintf("confluentKafkaProducer: Delivery failed: %v\n", ev.TopicPartition),
					zap.Any("tp", ev.TopicPartition))
				return
			}

			log.Debug(fmt.Sprintf("Successfully produced record to topic %s partition [%d] @ offset %v",
				*ev.TopicPartition.Topic, ev.TopicPartition.Partition, ev.TopicPartition.Offset),
				zap.Any("tp", ev.TopicPartition))

		case *kafka.Error:
			log.Error("confluentKafkaProducer: Received kafka.Error msg", zap.Error(ev))
		default:
			log.Error("confluentKafkaProducer: Received unknown error", zap.Any("ev", ev))
		}
	}
}