[KIP-932] Add error translation in share consumer by k-raina · Pull Request #2287 · confluentinc/confluent-kafka-python

Kaushik Raina (k-raina) · 2026-06-19T12:48:13Z

Why

The KIP-932 share consumer wrapped every librdkafka error in KafkaException, blurring client misuse (illegal state, concurrent access, bad args) and real broker errors.
This PR translates each error code to the exception, which align with java share consumer, i.e _STATE→IllegalStateException, _CONFLICT→ConcurrentModificationException, _INVALID_ARG→ValueError, everything else stays KafkaException — with the two new types added to cimpl and exported from confluent_kafka.

Notes / caveats

The new types subclass RuntimeError, not KafkaException, so except KafkaException no longer catches these for the share consumer. args[0] is still a KafkaError in every translated case, so .code()/.str() keep working.
Scope is ShareConsumer only — Producer/Consumer/Admin are unchanged.
The ack-commit callback and commit_sync's per-partition map intentionally keep KafkaException (broker results, matching Java's callback contract).

confluent-cla-assistant · 2026-06-19T12:48:24Z

🎉 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) · 2026-06-20T06:12:54Z

set_sasl_credentials was missed by the translation — it's still raising a bare RuntimeError on a closed handle here, and plain KafkaException via cfl_PyErr_from_error_destroy. Please change that as well.

Pranav Rathi (pranavrth) · 2026-06-20T06:22:08Z

I think we can create Pure python level exception instead of C Binding level and use it in C Binding. We should check the feasibility of that. Same we should do with Messages class.

Kaushik Raina (k-raina) · 2026-06-22T10:45:22Z

Thanks Pranav Rathi (@pranavrth) for reviews!

Regarding

I think we can create Pure python level exception instead of C Binding level and use it in C Binding. We should check the feasibility of that. Same we should do with Messages class.

Our C module cimpl loads first and error.py already does from confluent_kafka.cimpl import ....
If we move the exception classes into error.py and have cimpl import them while it's starting up, we get a cycle: cimpl needs error.py, but error.py needs cimpl. Since neither has finished loading, the import sees a half-built module and breaks.

The way around it is to not grab the Python classes at startup, but lazily at the first time C
actually needs to raise one and then cache the pointer so we only do it once. By then both modules are fully loaded, so there's no loop. It works fine, but note that: it adds a bit of C plumbing and a new edge case to get right.

Pranav Rathi (pranavrth) · 2026-06-22T13:05:29Z

Our C module cimpl loads first and error.py already does from confluent_kafka.cimpl import ....
If we move the exception classes into error.py and have cimpl import them while it's starting up, we get a cycle: cimpl needs error.py, but error.py needs cimpl. Since neither has finished loading, the import sees a half-built module and breaks.

The way around it is to not grab the Python classes at startup, but lazily at the first time C
actually needs to raise one and then cache the pointer so we only do it once. By then both modules are fully loaded, so there's no loop. It works fine, but note that: it adds a bit of C plumbing and a new edge case to get right.

We are fine with the current approach. No need to complicate things.

Pranav Rathi (pranavrth) · 2026-06-22T15:14:37Z

+            "closed.\n"
+            "\n"
+            "Subclass of RuntimeError. ``exception.args[0]`` is a "
+            ":py:class:`KafkaError`.\n"


These are not subtype of KafkaException so why do we keep KafkaError in these exceptions? We can keep this if its a big change right now but we don't need to keep KafkaError in these exceptions.

Pranav Rathi (pranavrth) · 2026-06-22T18:16:12Z

+                PyErr_SetObject(KafkaException, eo);
+                Py_DECREF(eo);
+        } else {
+                PyErr_SetString(etype, buf);


Suggested change

PyErr_SetString(etype, buf);

PyErr_SetString(etype, fmt ? buf : rd_kafka_err2str(code));

Pranav Rathi (pranavrth) · 2026-06-22T18:17:42Z

+        va_start(ap, fmt);
+        vsnprintf(buf, sizeof(buf), fmt, ap);
+        va_end(ap);


Suggested change

va_start(ap, fmt);

vsnprintf(buf, sizeof(buf), fmt, ap);

va_end(ap);

if(rmt) {

va_start(ap, fmt);

vsnprintf(buf, sizeof(buf), fmt, ap);

va_end(ap);

}

Pranav Rathi (pranavrth) · 2026-06-22T18:17:55Z

+ * a plain message string — the Python type already conveys the code, and these
+ * local errors have no retriable/fatal/abort flags worth preserving. */
+static void
+ShareConsumer_PyErr_Format(rd_kafka_resp_err_t code, const char *fmt, ...) {


Check KafkaError_new0

Pranav Rathi (pranavrth) · 2026-06-22T18:29:28Z

+                PyObject *eo = KafkaError_new0(code, "%s", buf);
+                if (!eo)
+                        return;
+                PyErr_SetObject(KafkaException, eo);
+                Py_DECREF(eo);


Use cfl_PyErr_Format instead.

Pranav Rathi (pranavrth) · 2026-06-22T18:38:41Z

+                ShareConsumer_PyErr_Format(RD_KAFKA_RESP_ERR__STATE, "%s",
+                                           ERR_MSG_SHARE_CONSUMER_CLOSED);


Suggested change

ShareConsumer_PyErr_Format(RD_KAFKA_RESP_ERR__STATE, "%s",

ERR_MSG_SHARE_CONSUMER_CLOSED);

PyErr_SetString(IllegalStateException,

ERR_MSG_SHARE_CONSUMER_CLOSED);

Same at other places

Pranav Rathi (pranavrth) · 2026-06-22T18:43:02Z

+# These subclass RuntimeError; str(exc) is the error message (not a KafkaError).
+class IllegalStateException(RuntimeError):
+    def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+    args: Tuple[Any, ...]
+
+class ConcurrentModificationException(RuntimeError):
+    def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+    args: Tuple[Any, ...]


Check if anything changed after we changed these errors to not take KafkaError

Pranav Rathi (pranavrth)

Build is failing. Check that.
Provided minor comment to be resolved in next PR.

Apart from that, LGTM!. Thanks Kaushik Raina (@k-raina) !.

Pranav Rathi (pranavrth) · 2026-06-22T23:33:40Z

+                if (!eo)
+                        return;


We should remove this as it is not setting error and returning as it is.

sonarqube-confluent · 2026-06-23T06:57:50Z

Quality Gate failed

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

See analysis details on SonarQube

Pranav Rathi (pranavrth)

LGTM!.

Kaushik Raina (k-raina) requested review from a team and Matthew Seal (MSeal) as code owners June 19, 2026 12:48

Add error translation in share consumer

da10f75

airlock-confluentinc Bot force-pushed the dev_kip-932_queues-for-kafka_error_translation branch from b2f55e4 to da10f75 Compare June 19, 2026 12:57

Kaushik Raina (k-raina) added 2 commits June 22, 2026 15:41

Update error translation for sasl share consumer API

9c42e42

Address comments

bcd7539

Fix CI

bf2fcce

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

minor

d57e98e

Pranav Rathi (pranavrth) requested changes Jun 22, 2026

View reviewed changes

Remove KafkaError from python genrated exceptions

bf73120

Pranav Rathi (pranavrth) requested changes Jun 22, 2026

View reviewed changes

Address comments

cabc920

Kaushik Raina (k-raina) requested a review from Pranav Rathi (pranavrth) June 22, 2026 19:51

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

View reviewed changes

Fix docs

364c852

address comments

d184f19

Kaushik Raina (k-raina) requested a review from Pranav Rathi (pranavrth) June 23, 2026 08:04

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

View reviewed changes

Kaushik Raina (k-raina) merged commit c0ef4b5 into dev_kip-932_queues-for-kafka Jun 23, 2026
2 checks passed

Kaushik Raina (k-raina) deleted the dev_kip-932_queues-for-kafka_error_translation branch June 23, 2026 08:44

	PyErr_SetString(etype, buf);
	PyErr_SetString(etype, fmt ? buf : rd_kafka_err2str(code));

		ShareConsumer_PyErr_Format(RD_KAFKA_RESP_ERR__STATE, "%s",
		ERR_MSG_SHARE_CONSUMER_CLOSED);

Uh oh!

Conversation

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

Uh oh!

Uh oh!

confluent-cla-assistant Bot commented Jun 19, 2026

Uh oh!

Pranav Rathi (pranavrth) commented Jun 20, 2026

Uh oh!

Pranav Rathi (pranavrth) commented Jun 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

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

Uh oh!

Pranav Rathi (pranavrth) commented Jun 22, 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!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

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!

sonarqube-confluent Bot commented Jun 23, 2026

Quality Gate failed

Uh oh!

Pranav Rathi (pranavrth) left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Kaushik Raina (k-raina) commented Jun 19, 2026 •

edited

Loading

Pranav Rathi (pranavrth) commented Jun 20, 2026 •

edited

Loading