Added topic id to describe group response (#1645)

Added topic id to describe group response

Author: pranavrth

docs/index.rst

@@ -30,6 +30,7 @@ Supporting classes
     - :ref:`Node <pythonclient_node>`
     - :ref:`ConsumerGroupTopicPartitions <pythonclient_consumer_group_topic_partition>`
     - :ref:`ConsumerGroupState <pythonclient_consumer_group_state>`
+    - :ref:`Uuid <pythonclient_uuid>`
 
     - Errors:
        - :ref:`KafkaError <pythonclient_kafkaerror>`
@@ -660,6 +661,15 @@ ConsumerGroupState
 .. autoclass:: confluent_kafka.ConsumerGroupState
    :members:
 
+.. _pythonclient_uuid:
+
+****
+Uuid
+****
+
+.. autoclass:: confluent_kafka.Uuid
+   :members:
+
 .. _serde_field:
 
 ************

examples/adminapi.py

@@ -540,6 +540,7 @@ def example_describe_topics(a, args):
         try:
             t = future.result()
             print("Topic name             : {}".format(t.name))
+            print("Topic id               : {}".format(t.topic_id))
             if (t.is_internal):
                 print("Topic is Internal")
 

src/confluent_kafka/__init__.py

@@ -29,6 +29,7 @@
                     Consumer,
                     Message,
                     TopicPartition,
+                    Uuid,
                     libversion,
                     version,
                     TIMESTAMP_NOT_AVAILABLE,
@@ -46,7 +47,7 @@
            'Producer', 'DeserializingConsumer',
            'SerializingProducer', 'TIMESTAMP_CREATE_TIME', 'TIMESTAMP_LOG_APPEND_TIME',
            'TIMESTAMP_NOT_AVAILABLE', 'TopicPartition', 'Node',
-           'ConsumerGroupTopicPartitions', 'ConsumerGroupState']
+           'ConsumerGroupTopicPartitions', 'ConsumerGroupState', 'Uuid']
 
 __version__ = version()[0]
 

src/confluent_kafka/admin/_topic.py

@@ -26,14 +26,19 @@ class TopicDescription:
     ----------
     name : str
         The topic name.
+    topic_id: Uuid
+        The topic id of the topic
+    is_internal:
+        Whether the topic is internal or not
     partitions : list(TopicPartitionInfo)
         Partition information.
     authorized_operations: list(AclOperation)
         AclOperations allowed for the topic.
     """
 
-    def __init__(self, name, is_internal, partitions, authorized_operations=None):
+    def __init__(self, name, topic_id, is_internal, partitions, authorized_operations=None):
         self.name = name
+        self.topic_id = topic_id
         self.is_internal = is_internal
         self.partitions = partitions
         self.authorized_operations = None

src/confluent_kafka/src/Admin.c

@@ -3819,6 +3819,7 @@ static PyObject *Admin_c_TopicDescription_to_py(
         size_t c_authorized_operations_cnt = 0;
         size_t i = 0;
         const rd_kafka_AclOperation_t *c_authorized_operations = NULL;
+        const rd_kafka_Uuid_t *c_topic_id = NULL;
 
         TopicDescription_type = cfl_PyObject_lookup("confluent_kafka.admin",
                                                     "TopicDescription");
@@ -3833,6 +3834,11 @@ static PyObject *Admin_c_TopicDescription_to_py(
                              "name",
                              rd_kafka_TopicDescription_name(c_topic_description));
 
+        c_topic_id = rd_kafka_TopicDescription_topic_id(c_topic_description);
+        PyDict_SetItemString(kwargs,
+                             "topic_id",
+                             c_Uuid_to_py(c_topic_id));
+
         is_internal = PyBool_FromLong(rd_kafka_TopicDescription_is_internal(c_topic_description));
         if(PyDict_SetItemString(kwargs, "is_internal", is_internal) == -1) {
                 goto err;

src/confluent_kafka/src/confluent_kafka.c

@@ -810,6 +810,158 @@ PyObject *Message_new0 (const Handle *handle, const rd_kafka_message_t *rkm) {
 
 
 
+/****************************************************************************
+ *
+ *
+ * Uuid
+ *
+ *
+ *
+ *
+ ****************************************************************************/
+static PyObject *Uuid_most_significant_bits (Uuid *self, PyObject *ignore) {
+        if(self->cUuid) {
+                return cfl_PyLong_FromLong(rd_kafka_Uuid_most_significant_bits(self->cUuid));
+        }
+        Py_RETURN_NONE;
+}
+
+static PyObject *Uuid_least_significant_bits (Uuid *self, PyObject *ignore) {
+        if(self->cUuid) {
+                return cfl_PyLong_FromLong(rd_kafka_Uuid_least_significant_bits(self->cUuid));
+        }
+        Py_RETURN_NONE;
+}
+
+static PyMethodDef Uuid_methods[] = {
+	{ "get_most_significant_bits", (PyCFunction)Uuid_most_significant_bits, METH_NOARGS,
+	  "  :returns: Most significant 64 bits of the 128 bits Uuid\n"
+	  "  :rtype: int\n"
+	  "\n"
+	},
+	{ "get_least_significant_bits", (PyCFunction)Uuid_least_significant_bits, METH_NOARGS,
+	  "  :returns: Least significant 64 bits of the 128 bits Uuid\n"
+	  "  :rtype: int\n"
+	  "\n"
+	},
+	{ NULL }
+};
+
+
+static PyObject *Uuid_str0 (Uuid *self) {
+        if(self->cUuid) {
+                const char *base64str = rd_kafka_Uuid_base64str(self->cUuid);
+                if(base64str)
+                        return cfl_PyUnistr(_FromString(base64str));
+        }
+        Py_RETURN_NONE;
+}
+
+static long Uuid_hash (Uuid *self) {
+        
+        return rd_kafka_Uuid_most_significant_bits(self->cUuid) ^ rd_kafka_Uuid_least_significant_bits(self->cUuid);
+}
+
+
+static PyObject *Uuid_new (PyTypeObject *type, PyObject *args,
+                               PyObject *kwargs) {
+        PyObject *self = type->tp_alloc(type, 1);
+        return self;
+}
+
+
+static int Uuid_init (PyObject *self0, PyObject *args,
+                          PyObject *kwargs) {
+        Uuid *self = (Uuid *)self0;
+        static char *kws[] = { "most_significant_bits",
+                               "least_significant_bits",
+                               NULL };
+        int64_t most_significant_bits;
+        int64_t least_significant_bits;
+
+        if (!PyArg_ParseTupleAndKeywords(args, kwargs, "LL", kws,
+                                         &most_significant_bits, 
+                                         &least_significant_bits))
+                return -1;
+
+        self->cUuid = rd_kafka_Uuid_new(most_significant_bits, least_significant_bits);
+
+        return 0;
+}
+
+static int Uuid_clear (Uuid *self) {
+	if (self->cUuid) {
+		rd_kafka_Uuid_destroy(self->cUuid);
+		self->cUuid = NULL;
+	}
+	return 0;
+}
+
+static int Uuid_traverse (Uuid *self,
+                        visitproc visit, void *arg) {
+        return 0;
+}
+
+static void Uuid_dealloc (Uuid *self) {
+        PyObject_GC_UnTrack(self);
+        Uuid_clear(self);
+
+        Py_TYPE(self)->tp_free((PyObject *)self);
+}
+
+PyTypeObject UuidType = {
+        PyVarObject_HEAD_INIT(NULL, 0)
+        "cimpl.Uuid",                           /* tp_name */
+        sizeof(Uuid),                           /* tp_basicsize */
+        0,                                      /* tp_itemsize */
+        (destructor)Uuid_dealloc,               /* tp_dealloc */
+        0,                                      /* tp_print */
+        0,                                      /* tp_getattr */
+        0,                                      /* tp_setattr */
+        0,                                      /* tp_compare */
+        (reprfunc)Uuid_str0,                    /* tp_repr */
+        0,                                      /* tp_as_number */
+        0,                                      /* tp_as_sequence */
+        0,                                      /* tp_as_mapping */
+        (hashfunc)Uuid_hash,                    /* tp_hash */
+        0,                                      /* tp_call */
+        0,                                      /* tp_str */
+        PyObject_GenericGetAttr,                /* tp_getattro */
+        0,                                      /* tp_setattro */
+        0,                                      /* tp_as_buffer */
+        Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE | 
+	Py_TPFLAGS_HAVE_GC,                     /* tp_flags */
+        "Generic Uuid. Being used in various identifiers including topic_id.\n"
+        "\n"
+        ".. py:function:: Uuid(most_significant_bits, least_significant_bits)\n"
+        "\n"
+        "  Instantiate a Uuid object.\n"
+        "\n"
+        "  :param long most_significant_bits: Most significant 64 bits of the 128 bits Uuid.\n"
+        "  :param long least_significant_bits: Least significant 64 bits of the 128 bits Uuid.\n"
+        "  :rtype: Uuid\n"
+        "\n"
+"\n",                                           /* tp_doc */
+        (traverseproc)Uuid_traverse,            /* tp_traverse */
+        (inquiry)Uuid_clear,                    /* tp_clear */
+        0,                                      /* tp_richcompare */
+        0,                                      /* tp_weaklistoffset */
+        0,                                      /* tp_iter */
+        0,                                      /* tp_iternext */
+        Uuid_methods,                           /* tp_methods */
+        0,                                      /* tp_members */
+        0,                                      /* tp_getset */
+        0,                                      /* tp_base */
+        0,                                      /* tp_dict */
+        0,                                      /* tp_descr_get */
+        0,                                      /* tp_descr_set */
+        0,                                      /* tp_dictoffset */
+        Uuid_init,                              /* tp_init */
+        0,                                      /* tp_alloc */
+        Uuid_new                                /* tp_new */
+};
+
+
 
 /****************************************************************************
  *
@@ -1488,6 +1640,47 @@ PyObject *c_Node_to_py(const rd_kafka_Node_t *c_node) {
         return NULL;
 }
 
+PyObject *c_Uuid_to_py(const rd_kafka_Uuid_t *c_uuid) {
+        PyObject *uuid = NULL;
+        PyObject *Uuid_type = NULL;
+        PyObject *args = NULL;
+        PyObject *kwargs = NULL;
+        
+        if(!c_uuid)
+                Py_RETURN_NONE;
+
+        Uuid_type = cfl_PyObject_lookup("confluent_kafka",
+                                        "Uuid");
+        if (!Uuid_type) {
+                goto err;
+        }
+
+        
+        kwargs = PyDict_New();
+
+        cfl_PyDict_SetLong(kwargs, "most_significant_bits", rd_kafka_Uuid_most_significant_bits(c_uuid));
+        cfl_PyDict_SetLong(kwargs, "least_significant_bits", rd_kafka_Uuid_least_significant_bits(c_uuid));
+
+        args = PyTuple_New(0);
+
+        uuid = PyObject_Call(Uuid_type, args, kwargs);
+
+        Py_DECREF(Uuid_type);
+        Py_DECREF(args);
+        Py_DECREF(kwargs);
+
+        return uuid;
+
+err:
+        Py_XDECREF(Uuid_type);
+        Py_XDECREF(args);
+        Py_XDECREF(kwargs);
+        Py_XDECREF(uuid);
+
+        return NULL;
+
+}
+
 
 /****************************************************************************
  *
@@ -2834,6 +3027,8 @@ static PyObject *_init_cimpl (void) {
 		return NULL;
 	if (PyType_Ready(&MessageType) < 0)
 		return NULL;
+	if (PyType_Ready(&UuidType) < 0)
+		return NULL;
 	if (PyType_Ready(&TopicPartitionType) < 0)
 		return NULL;
 	if (PyType_Ready(&ProducerType) < 0)
@@ -2863,6 +3058,9 @@ static PyObject *_init_cimpl (void) {
 	Py_INCREF(&MessageType);
 	PyModule_AddObject(m, "Message", (PyObject *)&MessageType);
 
+        Py_INCREF(&UuidType);
+        PyModule_AddObject(m, "Uuid", (PyObject *)&UuidType);
+
 	Py_INCREF(&TopicPartitionType);
 	PyModule_AddObject(m, "TopicPartition",
 			   (PyObject *)&TopicPartitionType);

src/confluent_kafka/src/confluent_kafka.h

@@ -384,6 +384,7 @@ rd_kafka_conf_t *common_conf_setup (rd_kafka_type_t ktype,
 				    PyObject *kwargs);
 PyObject *c_parts_to_py (const rd_kafka_topic_partition_list_t *c_parts);
 PyObject *c_Node_to_py(const rd_kafka_Node_t *c_node);
+PyObject *c_Uuid_to_py(const rd_kafka_Uuid_t *c_uuid);
 rd_kafka_topic_partition_list_t *py_to_c_parts (PyObject *plist);
 PyObject *list_topics (Handle *self, PyObject *args, PyObject *kwargs);
 PyObject *list_groups (Handle *self, PyObject *args, PyObject *kwargs);
@@ -493,6 +494,14 @@ typedef struct {
 extern PyTypeObject NewTopicType;
 
 
+typedef struct {
+        PyObject_HEAD
+		rd_kafka_Uuid_t *cUuid;
+} Uuid;
+
+extern PyTypeObject UuidType;
+
+
 typedef struct {
         PyObject_HEAD
         char *topic;