Added Message.latency() to retrieve the per-message produce latency

Author: edenhill

src/confluent_kafka/src/confluent_kafka.c

@@ -483,6 +483,12 @@ static PyObject *Message_timestamp (Message *self, PyObject *ignore) {
 			     self->timestamp);
 }
 
+static PyObject *Message_latency (Message *self, PyObject *ignore) {
+        if (self->latency == -1)
+                Py_RETURN_NONE;
+	return PyFloat_FromDouble((double)self->latency / 1000000.0);
+}
+
 static PyObject *Message_headers (Message *self, PyObject *ignore) {
 #ifdef RD_KAFKA_V_HEADERS
 	if (self->headers) {
@@ -587,6 +593,18 @@ static PyMethodDef Message_methods[] = {
 	  "  :rtype: (int, int)\n"
 	  "\n"
 	},
+	{ "latency", (PyCFunction)Message_latency, METH_NOARGS,
+          "Retrieve the time it took to produce the message, from calling "
+          "produce() to the time the acknowledgement was received from "
+          "the broker.\n"
+          "Must only be used with the producer for successfully produced "
+          "messages.\n"
+          "\n"
+	  "  :returns: latency as float seconds, or None if latency "
+          "information is not available (such as for errored messages).\n"
+	  "  :rtype: float or None\n"
+	  "\n"
+	},
 	{ "headers", (PyCFunction)Message_headers, METH_NOARGS,
       "  Retrieve the headers set on a message. Each header is a key value"
       "pair. Please note that header keys are ordered and can repeat.\n"
@@ -769,6 +787,11 @@ PyObject *Message_new0 (const Handle *handle, const rd_kafka_message_t *rkm) {
 
 	self->timestamp = rd_kafka_message_timestamp(rkm, &self->tstype);
 
+        if (handle->type == RD_KAFKA_PRODUCER)
+                self->latency = rd_kafka_message_latency(rkm);
+        else
+                self->latency = -1;
+
 	return (PyObject *)self;
 }
 

src/confluent_kafka/src/confluent_kafka.h

@@ -417,6 +417,7 @@ typedef struct {
 	int64_t offset;
 	int64_t timestamp;
 	rd_kafka_timestamp_type_t tstype;
+        int64_t latency;  /**< Producer: time it took to produce message */
 } Message;
 
 extern PyTypeObject MessageType;

tests/integration/integration_test.py

@@ -135,8 +135,9 @@ def _delivery(err, msg, silent=False):
             return 0
         else:
             if not silent:
-                print('Message delivered to %s [%s] at offset [%s]: %s' %
-                      (msg.topic(), msg.partition(), msg.offset(), msg.value()))
+                print('Message delivered to %s [%s] at offset [%s] in %.3fs: %s' %
+                      (msg.topic(), msg.partition(), msg.offset(),
+                       msg.latency(), msg.value()))
             return 1
 
     def delivery(self, err, msg):
@@ -145,8 +146,9 @@ def delivery(self, err, msg):
                   (msg.topic(), str(msg.partition()), err))
             return
         elif not self.silent:
-            print('Message delivered to %s [%s] at offset [%s]: %s' %
-                  (msg.topic(), msg.partition(), msg.offset(), msg.value()))
+            print('Message delivered to %s [%s] at offset [%s] in %.3fs: %s' %
+                  (msg.topic(), msg.partition(), msg.offset(),
+                   msg.latency(), msg.value()))
         self.msgs_delivered += 1
         self.bytes_delivered += len(msg)
 

tests/test_Producer.py

@@ -27,9 +27,10 @@ def test_basic_api():
     p.produce('mytopic', value='somedata', key='a key')
 
     def on_delivery(err, msg):
-        print('delivery', str)
+        print('delivery', err, msg)
         # Since there is no broker, produced messages should time out.
         assert err.code() == KafkaError._MSG_TIMED_OUT
+        print('message latency', msg.latency())
 
     p.produce(topic='another_topic', value='testing', partition=9,
               callback=on_delivery)