KIP-932: Add Share Consumer soak tests + high-throughput perf variant by Ankith-Confluent · Pull Request #2230 · confluentinc/confluent-kafka-python

Ankith L (Ankith-Confluent) · 2026-04-14T06:16:26Z

What

Add --share flag to soakclient.py that runs Producer + ShareConsumer
Implicit and Explicit
Adds ShareConsumer (KIP-932) coverage to the CKPy soak client and introduces a
higher-throughput soakclient_perf.py variant for the HT soak

Checklist

Contains customer facing changes? Including API/behavior changes
Did you add sufficient unit test and/or integration test coverage for this PR?
- If not, please explain why it is not required

References

JIRA:

Test & Review

Open questions / Follow-ups

confluent-cla-assistant · 2026-04-14T06:16:41Z

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

Kaushik Raina (k-raina) · 2026-05-18T05:40:43Z

Ankith L (@Ankith-Confluent)
Could you please rebase this branch?

sonarqube-confluent · 2026-05-20T13:29:36Z

Quality Gate failed

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

See analysis details on SonarQube

sonarqube-confluent · 2026-06-12T13:16:43Z

Quality Gate failed

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

See analysis details on SonarQube

sonarqube-confluent · 2026-06-23T17:56:06Z

Quality Gate passed

Issues
30 New issues
3 Fixed issues
0 Accepted issues

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

See analysis details on SonarQube

…tionsx

…oak client

…trategy

… sync and async commits

…shes

…ure compatibility with librdkafka v2.15.0-RC1

…ging in commit_sync

…consistency in logging

Kaushik Raina (k-raina)

Is there scope to common out code between soakclient.py and soakclient_perf.py?

Kaushik Raina (k-raina) · 2026-07-01T08:58:17Z

+
+        if enable_share:
+            if not HAS_SHARE_CONSUMER:
+                raise RuntimeError("ShareConsumer requested but not available in this " "confluent_kafka build.")


In case share consumer fails, should producer also shut down?

Kaushik Raina (k-raina) · 2026-07-01T09:01:35Z

+                            self.share_err_cnt += 1
+                            self.incr_counter("consumer.error", 1)
+                    else:
+                        self.share_consumer.commit_async()


Should we plan to add ack callback to catch errors?

Yes we can do that, when I implemented this before, that mechanism was not implemented yet in Librdkafka side.
Will add.

Kaushik Raina (k-raina) · 2026-07-01T09:11:28Z

+                    if hw > 0:
+                        if msg.offset() <= hw:
+                            self.logger.warning(
+                                "share: Old or duplicate message {} "


Does this logic prints "Old or duplicate message" for delivery retries?

yes, I will change the the log text.

Kaushik Raina (k-raina) · 2026-07-01T09:18:38Z

                self.msg_err_cnt += 1
                self.incr_counter("consumer.msgerr", 1)

            self.msg_cnt += 1


Do we want to count corrupt messages also?

I think we should not count them because we already have another metric counting the erros.
So it would pollute the readings.

Kaushik Raina (k-raina) · 2026-07-01T09:22:18Z

+                    self.share_err_cnt += 1
+                    self.incr_counter("consumer.error", 1)
+
+        self.share_consumer.close()


nit: should close be moved into finally?

Kaushik Raina (k-raina) · 2026-07-01T09:25:29Z

+            self.logger.info("share: aborted by user")
+            self.run = False
+        except Exception as ex:
+            self.logger.fatal("share: fatal exception: {}\n{}".format(ex, traceback.print_exc()))


Suggested change

self.logger.fatal("share: fatal exception: {}\n{}".format(ex, traceback.print_exc()))

self.logger.fatal("share: fatal exception: {}\n{}".format(ex, traceback.format_exc()))

Check if format_exec is more relavant? Same for rest of files

Kaushik Raina (k-raina) · 2026-07-01T09:41:08Z


        # Create topic (might already exist)
-        aconf = filter_config(conf, ["consumer.", "producer."], "admin.")
+        aconf = filter_config(conf, ["consumer.", "producer.", "share."], "admin.")


Adding ".share" in filter config changes share.acknowledgement.mode -> acknowledgement.mode, which is an invalid config. i think this will cause crash for explicit mode. Is my understanding correct?

The second arg to filter_config is the drop list, not a rename. So share.acknowledgement.mode is dropped from aconf, not renamed to acknowledgement.mode.
Explicit mode still works because sconf['share.acknowledgement.mode'] = 'explicit' is set directly on the share conf after filter_config
I verified it end-to-end with --share --explicit.

Ankith L (Ankith-Confluent) · 2026-07-01T10:29:29Z

Thanks! Kaushik Raina (@k-raina)
I have addressed the comments.
Regarding the point to common out the code, I think it's better to keep it separated and simple, in the future if we want to modify the perf script, we will be required to change the soak client as well.
What are your thoughts about this?

Kaushik Raina (k-raina) · 2026-07-01T11:51:17Z

In the current codebase, the two files (soakclient.py and soakclient_perf.py) overlap heavily:

Out of the 31 function definitions, 28 are byte-for-byte identical.
Only 3 functions differ in actual behavior: SoakRecord.init, SoakRecord.serialize, producer_run

Because the two files are largely copies, any fix applied to one must be manually ported to the other. Over time this leads to silent, unintended drift (for example, divergence in logging detail, error handing etc.).

One way to address this is to make soakclient.py the base class and soakclient_perf.py a subclass.

…kclient_perf.py

sonarqube-confluent · 2026-07-02T05:39:06Z

Quality Gate passed

Issues
13 New issues
6 Fixed issues
0 Accepted issues

Measures
1 Security Hotspot
No data about Coverage
0.0% Duplication on New Code

See analysis details on SonarQube

Kaushik Raina (k-raina)

Overall LGTM!
Since delta btw files is tiny, runtime branching works fine for this PR. In future if there are many hooks or variants, we might need to revisit and adopt Inheritance.

Kaushik Raina (k-raina) · 2026-07-03T10:55:54Z

        self.msg_err_cnt = 0
        self.consumer_err_cnt = 0
        self.consumer_error_cb_cnt = 0
        self.last_commited = None


Does this need to be?

Suggested change

self.last_committed = None

Kaushik Raina (k-raina) · 2026-07-03T10:59:00Z

            self.incr_counter("consumer.msg", 1)

            # end-to-end latency
            headers = dict(msg.headers())


Nit: Does this crash for msg with no header?

Kaushik Raina (k-raina) · 2026-07-03T11:02:27Z

+            # duration the OS can actually honor. -r remains the target rate
+            # (an upper bound: if a batch takes longer than its time budget,
+            # we run as fast as we can, below -r).
+            batch = max(1, int(self.rate / 100))  # ~100 batches/sec


Note : By default , rate is hardcoded to 80 in run.sh:

time opentelemetry-instrument $testdir/soakclient.py -i $TESTID -t $topic -r 80 -f $1 $share_flag $explicit_flag $perf_flag

So each batch is one message and batching only kicks in at -r ≥ 200.

Ankith L (Ankith-Confluent) requested review from a team and Matthew Seal (MSeal) as code owners April 14, 2026 06:16

Ankith L (Ankith-Confluent) changed the base branch from dev_kip-932_queues-for-kafka to dev_kip-932_share_consumer_poll April 27, 2026 10:01

Base automatically changed from dev_kip-932_share_consumer_poll to dev_kip-932_queues-for-kafka May 13, 2026 11:41

airlock-confluentinc Bot force-pushed the dev_kip-932_queues-for-kafka branch from a980019 to b5cb017 Compare May 20, 2026 05:58

airlock-confluentinc Bot force-pushed the dev_kip-932_soak_tests branch from 78301f5 to 8e9dcd0 Compare May 20, 2026 12:50

airlock-confluentinc Bot force-pushed the dev_kip-932_soak_tests branch from 2978328 to 2153fe2 Compare June 4, 2026 10:24

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_soak_tests branch from 770ec69 to 449ec68 Compare June 11, 2026 16:35

airlock-confluentinc Bot force-pushed the dev_kip-932_soak_tests branch from 2ad627d to 4f8327e Compare June 19, 2026 15:54

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 16:30

airlock-confluentinc Bot force-pushed the dev_kip-932_soak_tests branch from 65ff773 to 067298d Compare June 23, 2026 16:30

Ankith L (Ankith-Confluent) removed the request for review from Copilot June 23, 2026 16:30

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

Ankith L (Ankith-Confluent) removed the request for review from Copilot June 23, 2026 16:39

Copilot AI review requested due to automatic review settings June 23, 2026 16:58

Ankith L (Ankith-Confluent) removed the request for review from Copilot June 23, 2026 16:58

Ankith L (Ankith-Confluent) added 6 commits July 1, 2026 10:22

Add support for ShareConsumer in soak client and update usage instruc…

075d09c

…tionsx

Refactor share consumer metrics to use generic consumer counters in s…

18ca7ef

…oak client

Update share consumer method from consume_batch to poll in SoakClient

b816d84

Remove statistics configuration for ShareConsumer to prevent crashes

333fdf1

Add explicit mode support for ShareConsumer with alternating commit s…

d59a21a

…trategy

Refactor explicit mode handling in ShareConsumer to alternate between…

f2e37c8

… sync and async commits

Ankith L (Ankith-Confluent) added 2 commits July 1, 2026 10:22

Remove unused statistics configuration from SoakClient to prevent cra…

4a32ba1

…shes

Remove 'enable.partition.eof' from ShareConsumer configuration to ens…

b219457

…ure compatibility with librdkafka v2.15.0-RC1

Copilot AI review requested due to automatic review settings July 1, 2026 05:04

airlock-confluentinc Bot force-pushed the dev_kip-932_soak_tests branch from c2aad2f to b219457 Compare July 1, 2026 05:04

Ankith L (Ankith-Confluent) removed the request for review from Copilot July 1, 2026 05:04

Refactor SoakClient to streamline gauge setting and improve error log…

6533268

…ging in commit_sync

Copilot AI review requested due to automatic review settings July 1, 2026 05:05

Ankith L (Ankith-Confluent) removed the request for review from Copilot July 1, 2026 05:05

Validate SHARE flag when EXPLICIT mode is enabled in soak tests

71d53f6

Copilot AI review requested due to automatic review settings July 1, 2026 05:20

Ankith L (Ankith-Confluent) removed the request for review from Copilot July 1, 2026 05:20

Ankith L (Ankith-Confluent) changed the title ~~KIP-932: Add ShareConsumer support to soak tests~~ KIP-932: Add ShareConsumer soak tests + high-throughput perf variant Jul 1, 2026

Ankith L (Ankith-Confluent) changed the title ~~KIP-932: Add ShareConsumer soak tests + high-throughput perf variant~~ KIP-932: Add Share Consumer soak tests + high-throughput perf variant Jul 1, 2026

Refactor soakclient and soakclient_perf for improved readability and …

f2c1d28

…consistency in logging

Copilot AI review requested due to automatic review settings July 1, 2026 08:24

Ankith L (Ankith-Confluent) removed the request for review from Copilot July 1, 2026 08:24

Kaushik Raina (k-raina) reviewed Jul 1, 2026

View reviewed changes

Adress the comments

ab89543

Copilot AI review requested due to automatic review settings July 1, 2026 10:26

Ankith L (Ankith-Confluent) removed the request for review from Copilot July 1, 2026 10:26

Add --perf flag for high-throughput mode in soak tests and remove soa…

87bc749

…kclient_perf.py

Copilot AI review requested due to automatic review settings July 2, 2026 04:20

Ankith L (Ankith-Confluent) removed the request for review from Copilot July 2, 2026 04:20

Enforce SHARE flag requirement for PERF mode in soak tests

2f6e5ab

Copilot AI review requested due to automatic review settings July 2, 2026 05:01

Ankith L (Ankith-Confluent) removed the request for review from Copilot July 2, 2026 05:01

Kaushik Raina (k-raina) reviewed Jul 3, 2026

View reviewed changes

	self.logger.fatal("share: fatal exception: {}\n{}".format(ex, traceback.print_exc()))
	self.logger.fatal("share: fatal exception: {}\n{}".format(ex, traceback.format_exc()))

Uh oh!

Conversation

Ankith L (Ankith-Confluent) commented Apr 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Checklist

References

Test & Review

Open questions / Follow-ups

Uh oh!

confluent-cla-assistant Bot commented Apr 14, 2026

Uh oh!

Kaushik Raina (k-raina) commented May 18, 2026

Uh oh!

sonarqube-confluent Bot commented May 20, 2026

Quality Gate failed

Uh oh!

sonarqube-confluent Bot commented Jun 12, 2026

Quality Gate failed

Uh oh!

sonarqube-confluent Bot commented Jun 23, 2026

Quality Gate passed

Uh oh!

Kaushik Raina (k-raina) 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!

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!

Ankith L (Ankith-Confluent) commented Jul 1, 2026

Uh oh!

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

Uh oh!

Uh oh!

sonarqube-confluent Bot commented Jul 2, 2026

Quality Gate passed

Uh oh!

Kaushik Raina (k-raina) 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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Ankith L (Ankith-Confluent) commented Apr 14, 2026 •

edited

Loading

Kaushik Raina (k-raina) commented Jul 1, 2026 •

edited

Loading