CLIENTS-1765: Added support for consumer.memberid() #1154 (#1455)

Added support for consumer.memberid()

CHANGELOG.md

@@ -1,9 +1,11 @@
 # Confluent's Python client for Apache Kafka
 
 
-## v1.10.0
+## v1.9.3
 
-- Add metadata to TopicPartition type and commit() (#1410).
+ - Added metadata to `TopicPartition` type and `commit()` (#1410).
+ - Added `consumer.memberid()` for getting member id assigned to 
+   the consumer in a consumer group (#1154).
 
 
 ## v1.9.2

src/confluent_kafka/src/Consumer.c

@@ -980,6 +980,33 @@ static PyObject *Consumer_poll (Handle *self, PyObject *args,
 }
 
 
+static PyObject *Consumer_memberid (Handle *self, PyObject *args,
+                                    PyObject *kwargs) {
+        char *memberid;
+        PyObject *memberidobj;
+        if (!self->rk) {
+                PyErr_SetString(PyExc_RuntimeError,
+                                "Consumer closed");
+                return NULL;
+        }
+
+        memberid = rd_kafka_memberid(self->rk);
+
+        if (!memberid)
+                Py_RETURN_NONE;
+
+        if (!*memberid) {
+                rd_kafka_mem_free(self->rk, memberid);
+                Py_RETURN_NONE;
+        }
+
+        memberidobj = Py_BuildValue("s", memberid);
+        rd_kafka_mem_free(self->rk, memberid);
+
+        return memberidobj;
+}
+
+
 static PyObject *Consumer_consume (Handle *self, PyObject *args,
                                         PyObject *kwargs) {
         unsigned int num_messages = 1;
@@ -1409,6 +1436,19 @@ static PyMethodDef Consumer_methods[] = {
           "  :raises: RuntimeError if called on a closed consumer\n"
           "\n"
         },
+        { "memberid", (PyCFunction)Consumer_memberid, METH_NOARGS,
+          ".. py:function:: memberid()\n"
+          "\n"
+          " Return this client's broker-assigned group member id.\n"
+          "\n"
+          " The member id is assigned by the group coordinator and"
+          " is propagated to the consumer during rebalance.\n"
+          "\n"
+          "  :returns: Member id string or None\n"
+          "  :rtype: string\n"
+          "  :raises: RuntimeError if called on a closed consumer\n"
+          "\n"
+        },
 	{ "close", (PyCFunction)Consumer_close, METH_NOARGS,
 	  "\n"
 	  "  Close down and terminate the Kafka Consumer.\n"

tests/integration/consumer/test_consumer_memberid.py

@@ -0,0 +1,50 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Copyright 2022 Confluent Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limit
+
+import pytest
+
+
+def test_consumer_memberid(kafka_cluster):
+    """
+    Test consumer memberid.
+    """
+
+    consumer_conf = {'group.id': 'test'}
+
+    topic = "testmemberid"
+
+    kafka_cluster.create_topic(topic)
+
+    consumer = kafka_cluster.consumer(consumer_conf)
+
+    assert consumer is not None
+    assert consumer.memberid() is None
+    kafka_cluster.seed_topic(topic, value_source=[b'memberid'])
+
+    consumer.subscribe([topic])
+    msg = consumer.poll(10)
+    assert msg is not None
+    assert msg.value() == b'memberid'
+    memberid = consumer.memberid()
+    print("Member Id is -----> " + memberid)
+    assert isinstance(memberid, str)
+    assert len(memberid) > 0
+    consumer.close()
+
+    with pytest.raises(RuntimeError) as error_info:
+        consumer.memberid()
+    assert error_info.value.args[0] == "Consumer closed"