Support rd_kafka_memberid() from python clients

CHANGELOG.md

@@ -1,5 +1,18 @@
 # Confluent's Python client for Apache Kafka
 
+## v1.8.0
+
+v1.8.0 is a maintenance release with the following fixes and enhancements:
+
+### Enhancements
+
+- Support rd_kafka_memberid() from python clients (#1154).
+
+confluent-kafka-python is based on librdkafka v1.8.0, see the
+[librdkafka release notes](https://github.com/edenhill/librdkafka/releases/tag/v1.8.0)
+for a complete list of changes, enhancements, fixes and upgrade considerations.
+
+
 ## v1.7.0
 
 v1.7.0 is a maintenance release with the following fixes and enhancements:

src/confluent_kafka/src/Consumer.c

@@ -980,6 +980,28 @@ 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;
+
+        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 +1431,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,42 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Copyright 2021 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
+
+
+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 len(consumer.memberid()) == 0
+    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'
+    assert isinstance(consumer.memberid(), str) is True
+    assert len(consumer.memberid()) > 0
+    consumer.close()