[KIP-932] Add deserializing share consumer by k-raina · Pull Request #2265 · confluentinc/confluent-kafka-python

Kaushik Raina (k-raina) · 2026-06-04T14:44:27Z

Summary

Adds DeserializingShareConsumer, a high-level KIP-932 share consumer that deserializes record keys and values via the key.deserializer / value.deserializer config (same callables as DeserializingConsumer).
Overrides poll() to return a list of messages (mirroring ShareConsumer) and deserializes each one in place, so the returned messages stay valid arguments to acknowledge().
A per-record deserialization failure no longer drops the rest of the batch: the raw bytes are kept and msg.error() is set to KEY_DESERIALIZATION / _VALUE_DESERIALIZATION, so apps can detect it with the usual if msg.error(): check and acknowledge with REJECT.
Exports the class from confluent_kafka and adds the missing Message.set_error to the cimpl type stub.
Adds unit tests , an integration test, and supporting share-consumer test fixtures

confluent-cla-assistant · 2026-06-04T14:44:39Z

🎉 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.}

Pranav Rathi (pranavrth)

I have provided few comments. Please fix those and then I will review this PR again.

Pranav Rathi (pranavrth) · 2026-06-07T20:23:55Z

+        topic = msg.topic()
+        if topic is None:
+            # no topic to deserialize against; flag it rather than raise
+            msg.set_error(KafkaError(KafkaError._VALUE_DESERIALIZATION, "Message topic is None"))
+            return


This should be a TypeError. Check Deserializing Consumer and make sure we are following those conventions.

Pranav Rathi (pranavrth) · 2026-06-07T20:30:53Z

I think we should write integration tests with Avro, Protobuf and JSONSchema as well. A positive case and an error case with each is good enough to make sure that everything is fine.

Pranav Rathi (pranavrth) · 2026-06-07T20:44:51Z

+        if conf is not None:
+            share_conf.update(conf)
+
+        reset = share_conf.pop('auto.offset.reset', 'earliest')


This is incorrect use of the share consumer. Let's fix this even in above share consumer scenario and extract a common private function to reuse the same code.

Pranav Rathi (pranavrth) · 2026-06-07T20:46:47Z

    def set_headers(self, headers: HeadersType) -> None: ...
    def set_key(self, key: Any) -> None: ...
    def set_value(self, value: Any) -> None: ...
+    def set_error(self, error: KafkaError) -> None: ...


Do we really want to expose this function to the user? If yes then we have to add error handling in the case when the user is sending not error in the error field.

Kaushik Raina (k-raina) · 2026-06-11T12:04:16Z

Below are known test failures fixed in PR https://github.com/confluentinc/confluent-kafka-python/pull/2253/changes#top

1. test_subscribe_with_empty_list_raises
2. test_share_consumer_oauth_expired_token_surfaces_error_cb

Pranav Rathi (pranavrth) · 2026-06-11T19:29:31Z

+        if (new_error == Py_None) {
+                /* None clears the error; Message stores "no error" as NULL
+                 * (Message_error then returns None for it). */
+                self->error = NULL;


We never set error to NULL in python. It should be KafkaError or it should be None.

Pranav Rathi (pranavrth) · 2026-06-11T19:31:07Z

+        if (new_error != Py_None &&
+            !PyObject_TypeCheck(new_error, &KafkaErrorType)) {
+                PyErr_SetString(PyExc_TypeError,
+                                "error must be a KafkaError or None");
+                return NULL;
+        }


Add this breaking change in Changelog as well and add a TODO.

Pranav Rathi (pranavrth) · 2026-06-11T19:42:10Z

+
+        topic = msg.topic()
+        if topic is None:
+            raise TypeError("Message topic is None")


Add TODO to verify if this is correct or not for GA.

We have 3 options:

TypeError - but user didn't give the topic name so I think TypeError is not correct.

RuntimeError - I feel this is correct as this is internal issue not related to user.

KafkaError - set in the message field so that we don't throw exception.

Pranav Rathi (pranavrth) · 2026-06-11T19:49:37Z

+        """
+
+        messages = super(DeserializingShareConsumer, self).poll(timeout)
+        for msg in messages:


We need to check if we want to send error or exception in a message. We should use the pythonic way. Let's add a TODO to discuss it in the final interfaces part.

Pranav Rathi (pranavrth) · 2026-06-11T19:57:38Z

+            raise TypeError("Message topic is None")
+
+        try:
+            ctx = SerializationContext(topic, MessageField.VALUE, msg.headers())


Remove this to outside as it is being used below as well.

This is not done.

Pranav Rathi (pranavrth) · 2026-06-11T20:06:43Z

+        if topic is None:
+            raise TypeError("Message topic is None")
+
+        try:


Move try to after if self._value_deserializer is not None: part.

Pranav Rathi (pranavrth)

1 comment is missed.

Pranav Rathi (pranavrth) · 2026-06-12T08:32:22Z

+            raise TypeError("Message topic is None")
+
+        try:
+            ctx = SerializationContext(topic, MessageField.VALUE, msg.headers())


This is not done.

Pranav Rathi (pranavrth)

Looks good to me. I am approving the PR but I would prefer a pass from Robert Yokota (@rayokota) or somebody from the @confluentinc/data-governance team to review this as well.

sonarqube-confluent · 2026-06-12T10:42:35Z

Quality Gate failed

Failed conditions
9.3% 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 June 4, 2026 14:44

Kaushik Raina (k-raina) requested a review from Pranav Rathi (pranavrth) June 7, 2026 10:46

Pranav Rathi (pranavrth) requested changes Jun 7, 2026

View reviewed changes

Kaushik Raina (k-raina) changed the base branch from dev_kip-932_share_consumer_existing_cb_support to dev_kip-932_queues-for-kafka June 8, 2026 06:57

airlock-confluentinc Bot force-pushed the dev_kip-932_share_consumer_deserializing_consumer branch from fb01516 to 3765147 Compare June 8, 2026 07:04

airlock-confluentinc Bot force-pushed the dev_kip-932_queues-for-kafka branch from d600e92 to 5e41108 Compare June 9, 2026 09:55

airlock-confluentinc Bot force-pushed the dev_kip-932_share_consumer_deserializing_consumer branch from 3765147 to 7087ef4 Compare June 10, 2026 21:47

Kaushik Raina (k-raina) requested a review from Pranav Rathi (pranavrth) June 11, 2026 07:12

Pranav Rathi (pranavrth) requested changes Jun 11, 2026

View reviewed changes

Kaushik Raina (k-raina) requested a review from Pranav Rathi (pranavrth) June 12, 2026 06:58

Pranav Rathi (pranavrth) requested changes Jun 12, 2026

View reviewed changes

Kaushik Raina (k-raina) added 5 commits June 12, 2026 15:27

KIP-932 : Add deserializing share consumer

51d3710

Address feedbacks

31302f5

Address comments

3e96d89

Address feedback

b5e744b

Minor

2dce544

airlock-confluentinc Bot force-pushed the dev_kip-932_share_consumer_deserializing_consumer branch from 2fabcb1 to 2dce544 Compare June 12, 2026 09:59

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

Pranav Rathi (pranavrth) approved these changes Jun 12, 2026

View reviewed changes

Conversation

Kaushik Raina (k-raina) commented Jun 4, 2026

Uh oh!

confluent-cla-assistant Bot commented Jun 4, 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!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

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 Jun 11, 2026

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!

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!

Uh oh!

Pranav Rathi (pranavrth) left a comment

Choose a reason for hiding this comment

Uh oh!

sonarqube-confluent Bot commented Jun 12, 2026

Quality Gate failed

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants