[KIP-932] - Add usage examples for share consumer applications by k-raina · Pull Request #2288 · confluentinc/confluent-kafka-python

Kaushik Raina (k-raina) · 2026-06-19T12:58:24Z

Summary

Adds runnable KIP-932 ShareConsumer examples to examples/, covering both
acknowledgement modes and the common deserialization/lifecycle patterns, plus
a README section indexing them.

Examples

File	Demonstrates
`share_consumer.py`	Implicit ack (default mode) — records acked on the next `poll()`
`share_consumer_context_manager.py`	`with` statement usage so `close()` runs automatically on exit
`share_consumer_commit_sync.py`	Explicit per-record ACCEPT/RELEASE/REJECT, committed synchronously with per-partition results
`share_consumer_commit_async.py`	Explicit ack committed asynchronously, with broker outcomes via the ack-commit callback
`share_consumer_avro.py`	`DeserializingShareConsumer` over Avro/Schema Registry, with poison-record (deserialization failure)

confluent-cla-assistant · 2026-06-19T12:58:35Z

🎉 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-23T12:04:31Z

-            messages = sc.poll(timeout=1.0)  # returns a list (possibly empty)
+            try:
+                messages = sc.poll(timeout=1.0)  # a list, possibly empty
+            except KafkaException as e:


Add IllegalStateException and ConcurrentModificationException handling as well. Only in this file. We will ask user to refer this file for error handling.

Pranav Rathi (pranavrth) · 2026-06-23T12:07:45Z

+                    # A bad record. In implicit mode you can't ack it by hand
+                    # (acknowledge() is rejected); the library automatically
+                    # retries it (temporary errors) or discards it (permanent
+                    # errors) on the next poll — it is never accepted. Just log it.


Suggested change

# A bad record. In implicit mode you can't ack it by hand

# (acknowledge() is rejected); the library automatically

# retries it (temporary errors) or discards it (permanent

# errors) on the next poll — it is never accepted. Just log it.

# the records with msg.error() field set will be acknowledged internally with RELEASE for temporary errors and REJECT for permanent errors. Check [KIP](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=255070434#KIP932:QueuesforKafka-Handlingbadrecords) for more details.

Pranav Rathi (pranavrth) · 2026-06-23T12:08:24Z

+                    # errors) on the next poll — it is never accepted. Just log it.
                    sys.stderr.write('%% Error: %s\n' % msg.error())
                    continue
                sys.stderr.write(


Suggested change

sys.stderr.write(

sys.stdout.write(

Pranav Rathi (pranavrth) · 2026-06-23T12:11:53Z

+            try:
+                messages = sc.poll(timeout=1.0)  # a list, possibly empty
+            except KafkaException as e:
+                # Re-raise fatal errors; otherwise log and keep going.


Suggested change

# Re-raise fatal errors; otherwise log and keep going.

# The consumer should stop consuming after fatal error.

Pranav Rathi (pranavrth) · 2026-06-23T12:14:41Z

+                        # A record we received but can't decode. In explicit
+                        # mode we still have to ack it — REJECT it as poison so
+                        # it isn't redelivered.
+                        sc.acknowledge(msg, AcknowledgeType.REJECT)


Suggested change

# A record we received but can't decode. In explicit

# mode we still have to ack it — REJECT it as poison so

# it isn't redelivered.

sc.acknowledge(msg, AcknowledgeType.REJECT)

# A record we received but can't decode. In explicit

# mode we still have to ack it — RELEASE it so that

# other consumer can pick and redeliver for processing

# again. In implicit ack mode, we currently don't release

# the record in the Preview internally.

sc.acknowledge(msg, AcknowledgeType.RELEASE)

Pranav Rathi (pranavrth) · 2026-06-23T12:20:15Z

+                try:
+                    # Your processing goes here.
+                    print(msg.value())
+                except Exception as e:
+                    # Couldn't process it — RELEASE so it can be retried.
+                    sys.stderr.write('%% Processing failed: %s\n' % e)
+                    sc.acknowledge(msg, AcknowledgeType.RELEASE)
+                    continue


no need for try catch.

Pranav Rathi (pranavrth) · 2026-06-23T12:21:17Z

+                sys.stderr.write('%% Consumer error: %s\n' % e)
+                continue
+
+            for msg in messages:


Let's move this to the first try itself.

Pranav Rathi (pranavrth) · 2026-06-23T12:23:19Z

+                sc.acknowledge(msg, AcknowledgeType.ACCEPT)
+
+            # Flush the acks before the next poll().
+            sc.commit_async()


remove as not needed. Poll will handle.

Pranav Rathi (pranavrth) · 2026-06-23T12:26:34Z

+            while True:
+                try:
+                    messages = sc.poll(timeout=1.0)  # a list, possibly empty
+                except KafkaException as e:


add other errors.

Pranav Rathi (pranavrth) · 2026-06-23T12:26:57Z

+                for msg in messages:
+                    if msg.error():
+                        sys.stderr.write('%% Error: %s\n' % msg.error())
+                        continue
+                    print(msg.value())


same move above

sonarqube-confluent · 2026-06-24T13:56:21Z

Quality Gate passed

Issues
0 New issues
0 Fixed issues
0 Accepted issues

Measures
0 Security Hotspots
No data about Coverage
0.0% Duplication on New Code

See analysis details on SonarQube

Pranav Rathi (pranavrth) · 2026-06-24T09:37:20Z

+                sys.stdout.write(
                    '%% %s [%d] at offset %d with key %s:\n'
                    % (msg.topic(), msg.partition(), msg.offset(), str(msg.key()))
                )
                print(msg.value())


Some where it is sys.stdout and some where it is print(). We should be consistent. I think use print itself.

Pranav Rathi (pranavrth) · 2026-06-24T15:27:18Z

+                # block; results land in the callback.
+                sc.commit_async()
+            except KafkaException as e:
+                # Re-raise fatal errors; otherwise log and keep going.


Add other exception handling here itself.

Pranav Rathi (pranavrth) · 2026-06-24T15:28:40Z

+            for msg in messages:
+                err = msg.error()
+                if err is not None:
+                    if err.code() in (KafkaError._KEY_DESERIALIZATION, KafkaError._VALUE_DESERIALIZATION):
+                        # A record we received but can't decode. In explicit
+                        # mode we still have to ack it — RELEASE it so that
+                        # other consumer can pick and redeliver for processing
+                        # again. In implicit ack mode, we currently don't release
+                        # the record in the Preview internally.
+                        sc.acknowledge(msg, AcknowledgeType.RELEASE)
+                    else:
+                        # Any other flagged record is acked internally by the
+                        # library — see share_consumer.py for the error handling.
+                        sys.stderr.write('%% Error: %s\n' % err)
+                    continue
+
+                # value is already deserialized.
+                user = msg.value()
+                if user is not None:
+                    print(
+                        "User record {}: name: {}, favorite_number: {}, favorite_color: {}".format(
+                            msg.key(), user.name, user.favorite_number, user.favorite_color
+                        )
+                    )
+                sc.acknowledge(msg, AcknowledgeType.ACCEPT)


Let's move this to try block.

Pranav Rathi (pranavrth) · 2026-06-24T15:28:48Z

            for msg in messages:
                if msg.error():
-                    # Per-message errors are informational; log and keep
-                    # polling. Truly fatal errors are raised out of poll()
-                    # itself via the error_cb path.
+                    # The records with msg.error() field set will be acknowledged
+                    # internally with RELEASE for temporary errors and REJECT for
+                    # permanent errors. Check KIP for more details.
                    sys.stderr.write('%% Error: %s\n' % msg.error())
                    continue
-                sys.stderr.write(
+                sys.stdout.write(
                    '%% %s [%d] at offset %d with key %s:\n'
                    % (msg.topic(), msg.partition(), msg.offset(), str(msg.key()))
                )
                print(msg.value())


Let's move this to try block.

Pranav Rathi (pranavrth) · 2026-06-24T15:30:51Z

+    try:
+        while True:
+            try:
+                batch = sc.poll(timeout=1.0)  # a list, possibly empty


use messages instead of batch.

Pranav Rathi (pranavrth) · 2026-06-24T15:31:07Z

+            for msg in batch:
+                if msg.error():
+                    # No need to acknowledge a flagged record — the library
+                    # already handles it. Acking it yourself is redundant and
+                    # can override a permanent discard with a retry. Just log it.
+                    sys.stderr.write('%% Error: %s\n' % msg.error())
+                    continue
+
+                try:
+                    # Your processing goes here.
+                    print(msg.value())
+                except Exception as e:
+                    # Couldn't process it — RELEASE so it can be retried.
+                    sys.stderr.write('%% Processing failed: %s\n' % e)
+                    sc.acknowledge(msg, AcknowledgeType.RELEASE)
+                    continue
+
+                sc.acknowledge(msg, AcknowledgeType.ACCEPT)
+
+            if not batch:
+                continue
+
+            # Flush the acks before the next poll().
+            results = sc.commit_sync(timeout=10.0)
+            for topic_partition, err in results.items():
+                if err is not None:
+                    sys.stderr.write(
+                        '%% Commit failed for %s [%d]: %s\n' % (topic_partition.topic, topic_partition.partition, err)
+                    )


Move this to try block.

Pranav Rathi (pranavrth) · 2026-06-24T15:33:54Z

+                        # already handles it. Acking it yourself is redundant and
+                        # can override a permanent discard with a retry. Just log it.


Library already acknowledges the errored records internally exception for Deserialization error where it needs to be acknowledged explicitly from the application (Will be consistent in GA). You can choose to override the acknowledge for the errored offsets if required.

Pranav Rathi (pranavrth) · 2026-06-24T15:34:31Z

+    # ShareConsumer configuration.
+    # See https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md


Let's remove this.

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

Base automatically changed from dev_kip-932_queues-for-kafka to master June 23, 2026 08:53

Copilot AI review requested due to automatic review settings June 23, 2026 11:39

Kaushik Raina (k-raina) removed the request for review from Copilot June 23, 2026 11:39

Kaushik Raina (k-raina) added 4 commits June 23, 2026 17:42

[KIP-932] Add share consumer examples

73e6612

Update comments

551e96f

Add usage examples

ea17d77

Add with clause example for share consumer

924855a

Pranav Rathi (pranavrth) requested changes Jun 23, 2026

View reviewed changes

Address comments

cf488e4

Copilot AI review requested due to automatic review settings June 23, 2026 13:06

airlock-confluentinc Bot force-pushed the dev_kip-932_queues-for-kafka_usage_examples branch from 5a7277c to cf488e4 Compare June 23, 2026 13:06

Kaushik Raina (k-raina) removed the request for review from Copilot June 23, 2026 13:06

Remove additional files

ace03a2

Copilot AI review requested due to automatic review settings June 23, 2026 13:17

Kaushik Raina (k-raina) removed the request for review from Copilot June 23, 2026 13:17

Update exception handing in with claude

22a09b9

Copilot AI review requested due to automatic review settings June 23, 2026 13:38

Kaushik Raina (k-raina) requested review from Pranav Rathi (pranavrth) and removed request for Copilot June 23, 2026 13:38

Minor

3ae9a64

Copilot AI review requested due to automatic review settings June 24, 2026 13:20

Kaushik Raina (k-raina) removed the request for review from Copilot June 24, 2026 13:20

Pranav Rathi (pranavrth) requested changes Jun 24, 2026

View reviewed changes

	# Re-raise fatal errors; otherwise log and keep going.
	# The consumer should stop consuming after fatal error.

		# already handles it. Acking it yourself is redundant and
		# can override a permanent discard with a retry. Just log it.

		# ShareConsumer configuration.
		# See https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md

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!

Summary

Examples

Uh oh!

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

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

Quality Gate passed

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!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

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

edited

Loading