Delivery report from events channel does not have timestamp and headers

[jasimsulaiman]

Description

Delivery report of message from events channel, when using channel producer does not have timestamp or headers. This behavior is same for successfully delivered, as well as on timeout on failure to deliver to kafka.

Not having headers makes it very hard to identify which message failed.

How to reproduce

Send message with timestamp:

        TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
        Value:          out,
        Headers:        mapToHeaders(headers),
        Timestamp:      time.Date(2018, 1, 1, 1, 1, 1, 1, time.UTC),
        TimestampType:  kafka.TimestampCreateTime,
    }

producer.ProduceChannel() <- msg

Checklist

Please provide the following information:

• [x] confluent-kafka-go and librdkafka version (LibraryVersion()): confluent-kafka-go:v1.3.0, librdkafka-1.3.0-r0
• [x] Apache Kafka broker version: kafka_2.12-2.4.0
• [x] Client configuration: ConfigMap{...} kafka.ConfigMap{"bootstrap.servers": serverIp, "broker.version.fallback": "2.1.1", "message.timeout.ms": "1000", "api.version.request": true}
• [x] Operating system: alpine:3.11
• [x] Provide client logs (with "debug": ".." as necessary) code:

      for e := range producer.Events() {
          switch ev := e.(type) {
          case *kafka.Message:
              m := ev
              log.Print("mes t: ", m.Timestamp)
          default:
              log.Print("default", e)
          }
      }
  }()

  log: 2020/02/02 14:58:46 mes t: 0001-01-01 00:00:00 +0000 UTC

• [x] Critical: yes

[silencev]

I also set metadata in headers and would like to know the delivery result from headers. for the timestamp, I'm using it to calculate the delivery latency.

@edenhill I saw here to get timestamp if event type is RD_KAFKA_EVENT_FETCH just curious is it possible to get message timestamp back by calling rd_kafka_message_timestamp() when receive event RD_KAFKA_EVENT_DR?

[edenhill]

@silencev Sorry, timestamps and headers are not available in the delivery report. You can pass a user object of your choice with each produced message which you can retrieve in the delivery report, you can use that to measure latency, et.al.

In librdkafka there's a rd_kafka_message_latency() which can be used in delivery reports to get the total time from produce() to delivery, it could easily be exposed in the Go client as well.

[nandak522]

user object of your choice

@edenhill Can you explain what do you mean by "user object" ? Are you referring to Message.Opaque struct field ?

[AdrielVelazquez]

@nandak522 and @jasimsulaiman This is an old issue, but I added the producer latency into this PR.

https://github.com/confluentinc/confluent-kafka-go/pull/1076

Unless there's a better method of getting the ProducerLatency that can circumvent this.