KIP-932 : Implement Share consumer interface with poll API by k-raina · Pull Request #2217 · confluentinc/confluent-kafka-python

Kaushik Raina (k-raina) · 2026-03-12T17:24:53Z

Summary

Implements ShareConsumer, a new Kafka client type for KIP-932 share group consumption
Added New file: src/confluent_kafka/src/ShareConsumer.c
ShareConsumerHandle struct inheriting from Handle via first-member embedding — enables safe (Handle *)self casts and reuses all common callback/TLS infrastructure
subscribe(topics) / unsubscribe() / subscription() — topic subscription management via rd_kafka_share_* APIs
consume_batch(timeout=-1) — batch-only consumption with chunked polling for Ctrl+C interruptibility; uses - CallState_begin/end and check_signals_between_chunks for correct GIL and TLS lifecycle management
close() — graceful shutdown: attempts broker close then always destroys handle

Known limitations (TODOs in code)

Item	Blocked On
rd_kafka_set_log_queue()	Needs a specific `rd_kafka_share_set_log_queue()` wrapper to handle the share consumer handle.
OAuth Background Callbacks	Missing `rd_kafka_share_sasl_background_callbacks_enable()` export in the C API.
max_poll_records	Currently hardcoded to 10005; requires updating to the `librdkafka` double-pointer API for dynamic config reads.

Additional Changes

Additional commit bc6101a to build python binding with kip-932 branch for librdkafka. We can remove this commit in future before merging to master.

confluent-cla-assistant · 2026-03-12T17:25:06Z

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

Copilot

Pull request overview

This PR introduces a new ShareConsumer Python API backed by a C-extension wrapper around librdkafka’s KIP-932 Share Consumer functionality, and wires it into the package distribution and type hints.

Changes:

Add a new C-extension type cimpl.ShareConsumer with subscribe(), unsubscribe(), subscription(), consume_batch(), and close().
Export ShareConsumer through confluent_kafka.cimpl and confluent_kafka.__init__, and add corresponding type stubs.
Add unit tests for the new ShareConsumer API (skipped when ShareConsumer/librdkafka support is unavailable).

Reviewed changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
tests/test_ShareConsumer.py	Adds test coverage for ShareConsumer construction, subscription handling, consume_batch behavior, and close semantics (with skip guard).
src/confluent_kafka/src/ShareConsumer.c	Implements the new ShareConsumer C-extension type and its methods.
src/confluent_kafka/src/confluent_kafka.h	Declares `ShareConsumerType` for module initialization.
src/confluent_kafka/src/confluent_kafka.c	Registers the ShareConsumer Python type in the `cimpl` module init.
src/confluent_kafka/cimpl.pyi	Adds `ShareConsumer` type stubs.
src/confluent_kafka/init.py	Exposes `ShareConsumer` at the package top-level and in `__all__`.
setup.py	Adds `ShareConsumer.c` to the extension sources for building.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

You can also share your feedback on Copilot code review. Take the survey.

Copilot · 2026-03-12T17:32:03Z

src/confluent_kafka/src/ShareConsumer.c

+        self->base.type = RD_KAFKA_CONSUMER;
+
+        if (!(conf = common_conf_setup(RD_KAFKA_CONSUMER, (Handle *)self, args,
+                                       kwargs)))
+                return -1; /* Exception raised by common_conf_setup() */
+


ShareConsumer uses common_conf_setup(), which installs shared callbacks (error_cb/throttle_cb/stats_cb/logger/oauth). Those callbacks assume Handle.rk is a valid rd_kafka_t* and call rd_kafka_yield(h->rk) on callback crash/fatal paths. Since ShareConsumer never sets base.rk, a fatal error or a Python callback exception can end up calling rd_kafka_yield(NULL) and crash the process. Fix by either (a) updating the shared callbacks to yield using the rk argument they already receive (or rd_kafka_opaque(rk)), not h->rk, and using rk in oauth_cb as well, or (b) setting base.rk to an actual rd_kafka_t* for share consumers (if/when librdkafka exposes it).

Copilot · 2026-03-12T17:32:03Z

src/confluent_kafka/src/ShareConsumer.c

+        /* TODO: Remove after interface of librdkafka is updated to return double pointer */
+        self->max_poll_records = 10005;
+


max_poll_records is hardcoded to 10005 and consume_batch() allocates an array based on it on every call. This both diverges from the docstring (which says batch size is controlled by the max.poll.records config) and can cause unnecessary per-call allocation overhead. Suggest reading max.poll.records from the provided config dict during init (with a sane default matching librdkafka/Kafka, e.g. 500), validating bounds, and storing it in self->max_poll_records.

Suggested change

/* TODO: Remove after interface of librdkafka is updated to return double pointer */

self->max_poll_records = 10005;

/* Initialize max_poll_records from config (if provided), else use sane default. */

self->max_poll_records = 500; /* default, aligned with typical max.poll.records */

if (args && PyTuple_Check(args) && PyTuple_Size(args) >= 1) {

PyObject *conf_dict = PyTuple_GetItem(args, 0); /* borrowed ref */

if (conf_dict && PyDict_Check(conf_dict)) {

PyObject *mpr_obj =

PyDict_GetItemString(conf_dict, "max.poll.records"); /* borrowed */

if (mpr_obj) {

long mpr_val = PyLong_AsLong(mpr_obj);

if (mpr_val == -1 && PyErr_Occurred()) {

/* Non-integer or conversion error */

PyErr_SetString(PyExc_TypeError,

"max.poll.records must be an integer");

rd_kafka_conf_destroy(conf);

return -1;

}

/* Validate bounds: require a positive, reasonably bounded value. */

if (mpr_val < 1 || mpr_val > 1000000) {

PyErr_SetString(

PyExc_ValueError,

"max.poll.records must be between 1 and 1000000");

rd_kafka_conf_destroy(conf);

return -1;

}

self->max_poll_records = (size_t)mpr_val;

}

}

}

Copilot · 2026-03-12T17:32:03Z

src/confluent_kafka/src/ShareConsumer.c

+        CallState_begin((Handle *)self, &cs);
+        err = rd_kafka_share_consumer_close(self->rkshare);
+        CallState_end((Handle *)self, &cs);
+
+        /* Always destroy handle, to ensures resources are freed even if broker communication fails. */
+        CallState_begin((Handle *)self, &cs);
+        rd_kafka_share_destroy(self->rkshare);
+        CallState_end((Handle *)self, &cs);
+


ShareConsumer_close() ignores the return value of CallState_end(). If a Python signal is raised or a callback crashes during rd_kafka_share_consumer_close()/rd_kafka_share_destroy(), CallState_end() will return 0 and set an exception, but this code will continue and may mask the error or run with an inconsistent CallState. Mirror Consumer_close() by checking CallState_end() after each CallState_begin()/external call and returning NULL immediately on failure.

Copilot · 2026-03-12T17:32:04Z

src/confluent_kafka/src/ShareConsumer.c

+        if (!CallState_end((Handle *)self, &cs)) {
+                free(rkmessages);
+                if (error)
+                        rd_kafka_error_destroy(error);
+                return NULL;
+        }


If CallState_end() fails in consume_batch(), the code frees the rkmessages pointer array (and destroys error), but it does not destroy any messages that may already have been returned into rkmessages (rkmessages_size > 0). This leaks rd_kafka_message_t objects and differs from Consumer_consume(), which destroys messages before returning NULL. Ensure any populated messages are destroyed on this failure path before freeing the array.

Copilot · 2026-03-12T17:32:04Z

src/confluent_kafka/src/ShareConsumer.c

+     "  This method should be called to properly clean up the share consumer\n"
+     "  and leave the share group.\n"
+     "\n"
+     "  :raises RuntimeError: on error\n"


The close() docstring says it raises RuntimeError on error, but the implementation uses cfl_PyErr_Format(err, ...) which raises KafkaException (with the underlying rd_kafka_resp_err_t) on non-zero err. Please align the docstring with actual behavior (or change the exception type if RuntimeError is intended).

Suggested change

" :raises RuntimeError: on error\n"

" :raises KafkaException: on error\n"

Pratyush Ranjan (PratRanj07)

First pass comments

Pratyush Ranjan (PratRanj07) · 2026-03-15T21:05:28Z

src/confluent_kafka/src/ShareConsumer.c

+        rd_kafka_share_t *rkshare;
+
+        /* TODO: Remove after interface of librdkafka is updated to return double pointer */
+        size_t max_poll_records;


Lets name it batch_size so as to not confuse with the actual configuration property max.poll.records

Pratyush Ranjan (PratRanj07) · 2026-03-15T21:47:54Z

src/confluent_kafka/src/ShareConsumer.c

+    0,                                                    /*tp_as_buffer*/
+    Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE |
+        Py_TPFLAGS_HAVE_GC, /*tp_flags*/
+    "A high-level Apache Kafka share consumer (KIP-932)\n"


Suggested change

"A high-level Apache Kafka share consumer (KIP-932)\n"

"A high-level Apache Kafka share consumer\n"

Pratyush Ranjan (PratRanj07) · 2026-03-16T11:04:19Z

src/confluent_kafka/cimpl.pyi

+    def subscribe(self, topics: List[str]) -> None: ...
+    def unsubscribe(self) -> None: ...
+    def subscription(self) -> List[str]: ...
+    def consume_batch(self, timeout: float = -1) -> List[Message]: ...


I think we should name this Poll only to match the signature of the Java Client atleast in Higher level clients

In past, It was discussed that we would align naming with librdkafka APis.

I think we still need to finalize this name. Mostly we will go with poll to adhere to Java. It was mentioned in one of the recent comments in the librdkafka public interface docs.

Pratyush Ranjan (PratRanj07) · 2026-03-16T11:32:23Z

src/confluent_kafka/src/ShareConsumer.c

+ *
+ */
+static PyObject *
+ShareConsumer_consume_batch(ShareConsumerHandle *self,


Reduce the cyclomatic complexity of this function

This function definition is aligned with Consumer.consume_batch for simplicity

sonarqube-confluent · 2026-03-18T14:40:11Z

Quality Gate failed

Failed conditions
26.9% Coverage on New Code (required ≥ 80%)

See analysis details on SonarQube

Pranav Rathi (pranavrth) · 2026-04-02T05:55:44Z

I feel there will be issues related to the common callbacks in Python client. We are populating Handle but rk in that handle is not present. I can see that we are using this rk in the callbacks like we are using rd_kafka_yield or rd_kafka_fatal_error in error_cb which is using rk. The main decision in Python client is going to be related to reusing existing callback or Handle or devise a new code flow for the same. Let's go deeper in each callback atleast to figure out:

How are we planning to use Handle?
How are we panning to use internal rk handle? Some places are even destroying this handle with rd_kafka_destroy? Do we need to expose rk from librdkafka for Binding use?
Do we need to expose some extra APIs to make Python client work? Some of the APIs might not even work as those were working in normal Consumer. For example, rd_kafka_yield.
Verify each aspect of "ShareConsumerHandle struct inheriting from Handle via first-member embedding — enables safe (Handle *)self casts and reuses all common callback/TLS infrastructure".
Let's have a thorough understanding of how all the callbacks will work

Pranav Rathi (pranavrth)

Some high level comments. Still reviewing deeper.

Pranav Rathi (pranavrth) · 2026-04-02T00:35:53Z

src/confluent_kafka/cimpl.pyi

+    def subscribe(self, topics: List[str]) -> None: ...
+    def unsubscribe(self) -> None: ...
+    def subscription(self) -> List[str]: ...
+    def consume_batch(self, timeout: float = -1) -> List[Message]: ...


I think we still need to finalize this name. Mostly we will go with poll to adhere to Java. It was mentioned in one of the recent comments in the librdkafka public interface docs.

Pranav Rathi (pranavrth) · 2026-04-02T06:23:15Z

src/confluent_kafka/src/ShareConsumer.c

+                int is_fatal          = rd_kafka_error_is_fatal(error);
+                int is_retriable      = rd_kafka_error_is_retriable(error);


We are not planning to fill this information judiciously. Let's not use this.

Pranav Rathi (pranavrth) · 2026-04-02T06:27:57Z

src/confluent_kafka/src/ShareConsumer.c

+                Py_RETURN_NONE;
+
+        CallState_begin((Handle *)self, &cs);
+        err = rd_kafka_share_consumer_close(self->rkshare);


return type of close is going to change to error_t *. Add a TODO to change this when interface in librdkafka is changed.

rd_kafka_error_t *rd_kafka_share_consumer_close(rd_kafka_share_t *rkshare);

Pranav Rathi (pranavrth) · 2026-04-06T11:41:27Z

src/confluent_kafka/src/ShareConsumer.c

+     "  :raises KafkaException: on error\n"
+     "\n"},
+
+    {NULL}};


We will need to add set_sasl_credentials and Context manager functions.

Pranav Rathi (pranavrth) · 2026-04-06T11:48:29Z

src/confluent_kafka/src/ShareConsumer.c

+        if (self->rkshare) {
+                CallState cs;
+                CallState_begin((Handle *)self, &cs);
+                rd_kafka_share_destroy(self->rkshare);


Use _flags variant for _destroy() here. If you cant add right now, add a TODO with TODO KIP-932.

Pranav Rathi (pranavrth) · 2026-04-06T12:00:47Z

src/confluent_kafka/src/ShareConsumer.c

+                if (is_fatal) {
+                        PyErr_Format(PyExc_RuntimeError, "Fatal error: %s",
+                                     error_str);
+                } else {
+                        PyErr_Format(KafkaException,
+                                     "Error: %s (retriable: %s)", error_str,
+                                     is_retriable ? "yes" : "no");
+                }
+


Check how Consumer is raising the exception. Use that way instead. cfl_PyErr_Format

Kaushik Raina (k-raina) · 2026-04-08T11:00:09Z

Thanks Pranav Rathi (@pranavrth) for review

ShareConsumerHandle embeds Handle base as its first member, so (Handle *)self casts are valid per the C standard.
- The fields actually used by ShareConsumer are: tlskey, error_cb, stats_cb, throttle_cb.
- Fields not used: base.rk (left NULL), base.u.Consumer.*.
When rd_kafka_share_consumer_new(conf) is called, it internally creates an rd_kafka_t* stored inside the opaque rd_kafka_share_t. The conf goes into this internal rk. So when callbacks fire, librdkafka passes this valid internal rk as the callback's first parameter.
- error_cb, stats_cb , throttle_cb, log_cb - uses the callback's rk param (valid), not h->rk. So these are safe.
- oauth_cb - will SEGFAULT. There are multiple h->rk usages that pass NULL to APIs that dereference it .eg. rk->rk_sasl.handle inside librdkafka
Regarding yield and destry functions:
- rd_kafka_yield(rk) — This function never dereferences its rk parameter — it only sets rd_kafka_yield_thread = 1 . So rd_kafka_yield(NULL) is safe.
- rd_kafka_destroy(rk) — Would segfault with NULL. Not currently reachable for ShareConsumer but would be a problem if OAuth support is added naively with wait_for_oauth_token_set.
The cleanest fix is a public accessor in librdkafka:
- rd_kafka_t *rd_kafka_share_get_rk(rd_kafka_share_t *rkshare); : Then ShareConsumer_init can do self->base.rk = rd_kafka_share_get_rk(self->rkshare), and all existing callback code, rd_kafka_set_log_queue(), and rd_kafka_sasl_background_callbacks_enable() work unmodified.

sonarqube-confluent · 2026-04-10T12:52:57Z

Quality Gate failed

Failed conditions
29.2% Coverage on New Code (required ≥ 80%)

See analysis details on SonarQube

Kaushik Raina (k-raina) requested review from a team and Matthew Seal (MSeal) as code owners March 12, 2026 17:24

Copilot AI review requested due to automatic review settings March 12, 2026 17:24

Copilot started reviewing on behalf of Kaushik Raina (k-raina) March 12, 2026 17:25 View session

Copilot AI reviewed Mar 12, 2026

View reviewed changes

Kaushik Raina (k-raina) changed the base branch from master to dev_kip-932_queues-for-kafka March 13, 2026 06:13

KIP-932 : Implement Share consumer interface with poll API

51ec85d

airlock-confluentinc bot force-pushed the dev_kip-932_share_consumer_poll branch 3 times, most recently from a7d9785 to bc6101a Compare March 16, 2026 06:12

BUILD-KIP932: Commit to build branch with kip-932 librdkafka branch

fc5ef4f

airlock-confluentinc bot force-pushed the dev_kip-932_share_consumer_poll branch from bc6101a to fc5ef4f Compare March 16, 2026 07:02

Pratyush Ranjan (PratRanj07) requested changes Mar 16, 2026

View reviewed changes

Address copilor comments

e23df2e

Kaushik Raina (k-raina) requested a review from Pratyush Ranjan (PratRanj07) March 18, 2026 05:04

Kaushik Raina (k-raina) added 2 commits March 18, 2026 11:14

Address first pass feedback

88b4a3a

Style fix

852d090

Pranav Rathi (pranavrth) requested changes Apr 6, 2026

View reviewed changes

Style fix

d59b196

Address feedback

48208f4

Kaushik Raina (k-raina) requested a review from Pranav Rathi (pranavrth) April 10, 2026 12:50

		/* TODO: Remove after interface of librdkafka is updated to return double pointer */
		self->max_poll_records = 10005;

-        /* TODO: Remove after interface of librdkafka is updated to return double pointer */
-        self->max_poll_records = 10005;
+        /* Initialize max_poll_records from config (if provided), else use sane default. */
+        self->max_poll_records = 500;  /* default, aligned with typical max.poll.records */
+        if (args && PyTuple_Check(args) && PyTuple_Size(args) >= 1) {
+                PyObject *conf_dict = PyTuple_GetItem(args, 0); /* borrowed ref */
+                if (conf_dict && PyDict_Check(conf_dict)) {
+                        PyObject *mpr_obj =
+                                PyDict_GetItemString(conf_dict, "max.poll.records"); /* borrowed */
+                        if (mpr_obj) {
+                                long mpr_val = PyLong_AsLong(mpr_obj);
+                                if (mpr_val == -1 && PyErr_Occurred()) {
+                                        /* Non-integer or conversion error */
+                                        PyErr_SetString(PyExc_TypeError,
+                                                        "max.poll.records must be an integer");
+                                        rd_kafka_conf_destroy(conf);
+                                        return -1;
+                                }
+                                /* Validate bounds: require a positive, reasonably bounded value. */
+                                if (mpr_val < 1 || mpr_val > 1000000) {
+                                        PyErr_SetString(
+                                            PyExc_ValueError,
+                                            "max.poll.records must be between 1 and 1000000");
+                                        rd_kafka_conf_destroy(conf);
+                                        return -1;
+                                }
+                                self->max_poll_records = (size_t)mpr_val;
+                        }
+                }
+        }

	" :raises RuntimeError: on error\n"
	" :raises KafkaException: on error\n"

	"A high-level Apache Kafka share consumer (KIP-932)\n"
	"A high-level Apache Kafka share consumer\n"

		int is_fatal = rd_kafka_error_is_fatal(error);
		int is_retriable = rd_kafka_error_is_retriable(error);

Conversation

Kaushik Raina (k-raina) commented Mar 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

confluent-cla-assistant bot commented Mar 12, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Mar 12, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 12, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 12, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 12, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 12, 2026

Choose a reason for hiding this comment

Uh oh!

Pratyush Ranjan (PratRanj07) left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sonarqube-confluent bot commented Mar 18, 2026

Quality Gate failed

Uh oh!

Pranav Rathi (pranavrth) commented Apr 2, 2026

Uh oh!

Pranav Rathi (pranavrth) left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Kaushik Raina (k-raina) commented Apr 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

sonarqube-confluent bot commented Apr 10, 2026

Quality Gate failed

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Kaushik Raina (k-raina) commented Mar 12, 2026 •

edited

Loading

Kaushik Raina (k-raina) commented Apr 8, 2026 •

edited

Loading