Fixed a segfault when 'commit' or 'store_offsets' consumer method is called incorrectly with errored Message object

Author: pranavrth

CHANGELOG.md

@@ -6,6 +6,7 @@ v2.4.1 is a maintenance release with the following fixes and enhancements:
 
  - Removed usage of `strcpy` to enhance security of the client (#1745)
  - Fixed invalid write in `OAUTHBEARER/OIDC` extensions copy (#1745)
+ - Fixed a segfault when `commit` or `store_offsets` consumer method is called incorrectly with errored Message object
 
 confluent-kafka-python is based on librdkafka v2.4.1, see the
 [librdkafka release notes](https://github.com/confluentinc/librdkafka/releases/tag/v2.4.1)

src/confluent_kafka/src/Consumer.c

@@ -496,6 +496,16 @@ static PyObject *Consumer_commit (Handle *self, PyObject *args,
 		}
 
 		m = (Message *)msg;
+                
+                if (m->error != Py_None) {
+                        PyObject *error = Message_error(m, NULL);
+                        PyObject *errstr = PyObject_CallMethod(error, "str", NULL);
+                        cfl_PyErr_Format(RD_KAFKA_RESP_ERR__INVALID_ARG,
+                                         "Cannot commit offsets for message with error: '%s'" , PyUnicode_AsUTF8(errstr));
+                        Py_DECREF(error);
+                        Py_DECREF(errstr);
+                        return NULL;
+                }
 
 		c_offsets = rd_kafka_topic_partition_list_new(1);
 		rktpar = rd_kafka_topic_partition_list_add(
@@ -627,6 +637,16 @@ static PyObject *Consumer_store_offsets (Handle *self, PyObject *args,
 
 		m = (Message *)msg;
 
+                if (m->error != Py_None) {
+                        PyObject *error = Message_error(m, NULL);
+                        PyObject *errstr = PyObject_CallMethod(error, "str", NULL);
+                        cfl_PyErr_Format(RD_KAFKA_RESP_ERR__INVALID_ARG,
+                                         "Cannot store offsets for message with error: '%s'" , PyUnicode_AsUTF8(errstr));
+                        Py_DECREF(error);
+                        Py_DECREF(errstr);
+                        return NULL;
+                }
+
 		c_offsets = rd_kafka_topic_partition_list_new(1);
 		rktpar = rd_kafka_topic_partition_list_add(
 			c_offsets, cfl_PyUnistr_AsUTF8(m->topic, &uo8),