[KIP-932] Add rd_kafka_share_commit_sync() API by pranavrth · Pull Request #5374 · confluentinc/librdkafka

Pranav Rathi (pranavrth) · 2026-03-30T13:01:01Z

Core implementation:
- SHARE_COMMIT_SYNC_FANOUT op type and handler in main thread
- Segregate acks by partition leader into per-broker pending_commit_sync
- Dispatch sync ack ops with priority over async_ack_details
- Timeout timer fills remaining partitions with _TIMED_OUT
- App thread blocks on temp queue until all broker replies or timeout
Data structures:
- commit_sync_request in rkcg (abs_timeout, results, wait count, replyq, timer)
- pending_commit_sync in broker (sync_ack_details, abs_timeout, request_id)
- SHARE_COMMIT_SYNC_FANOUT and FANOUT_REPLY op types with union members
Broker thread changes:
- ShareAcknowledge reply handler populates ack_results in reply op
- Main thread reply handler copies per-partition errors to sync results
- Pending sync acks dispatched before async when broker becomes free
Tests (0176-share_consumer_commit_sync.c):
- Basic implicit/explicit ack mode commit_sync
- No pending acks returns NULL/NULL
- Commit prevents redelivery (Consumer B gets 0 records)
- Mixed ack types (ACCEPT/RELEASE/REJECT) with delivery count verification
- Multiple commit_sync calls with back-to-back no-op commits
- Multi-topic multi-partition (10 topics x 6 partitions x 10 msgs)
- Mock: verifies ShareAcknowledge RPC count matches commit_sync calls
- Mock: timeout handling with 5s RTT / 2s timeout and recovery
- Mixed commit types (10 async + 1 sync pattern)
- Mock: broker dispatch priority (sync before async)
Example program and build integration

confluent-cla-assistant · 2026-03-30T13:01:12Z

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

1. Core implementation: - SHARE_COMMIT_SYNC_FANOUT op type and handler in main thread - Segregate acks by partition leader into per-broker pending_commit_sync - Dispatch sync ack ops with priority over async_ack_details - Timeout timer fills remaining partitions with _TIMED_OUT - App thread blocks on temp queue until all broker replies or timeout 2. Data structures: - commit_sync_request in rkcg (abs_timeout, results, wait count, replyq, timer) - pending_commit_sync in broker (sync_ack_details, abs_timeout, request_id) - SHARE_COMMIT_SYNC_FANOUT and FANOUT_REPLY op types with union members 3. Broker thread changes: - ShareAcknowledge reply handler populates ack_results in reply op - Main thread reply handler copies per-partition errors to sync results - Pending sync acks dispatched before async when broker becomes free 4. Tests (0176-share_consumer_commit_sync.c): - Basic implicit/explicit ack mode commit_sync - No pending acks returns NULL/NULL - Commit prevents redelivery (Consumer B gets 0 records) - Mixed ack types (ACCEPT/RELEASE/REJECT) with delivery count verification - Multiple commit_sync calls with back-to-back no-op commits - Multi-topic multi-partition (10 topics x 6 partitions x 10 msgs) - Mock: verifies ShareAcknowledge RPC count matches commit_sync calls - Mock: timeout handling with 5s RTT / 2s timeout and recovery - Mixed commit types (10 async + 1 sync pattern) - Mock: broker dispatch priority (sync before async) 5. Example program and build integration

1. Fix intermittent do_test_mixed_commit_types failure: add rd_sleep(3) before consumer close to allow in-flight async ack requests to complete. The race occurred when commit_async grabbed acks from the inflight map but the broker response hadn't arrived before consumer close tore down the connection. Tighten commit_sync timing assertion from 5000ms to 500ms 2. Add debug logging to commit_async, ShareAcknowledge request construction, and ack batch building for better observability 3. Add TODO KIP-932 comments for future improvements: - Ack batch merging for adjacent/overlapping offset ranges - Using abs_timeout instead of socket.timeout.ms for ShareAcknowledge requests in commit_sync path - Thread safety check for share fetch broker fields

Pratyush Ranjan (PratRanj07)

Some minor comments

Pratyush Ranjan (PratRanj07) · 2026-04-06T17:06:44Z

src/rdkafka.c

+
+        if (!ack_batches) {
+                rd_kafka_dbg(rk, CGRP, "SHARE",
+                             "No pending acknowledgements to commit sync");


Suggested change

"No pending acknowledgements to commit sync");

"No pending acknowledgements to commit");

Pratyush Ranjan (PratRanj07) · 2026-04-06T17:25:24Z

src/rdkafka.c

+            rkcg->rkcg_share.commit_sync_id_counter;
+        rkcg->rkcg_commit_sync_request.abs_timeout = abs_timeout;
+        rkcg->rkcg_commit_sync_request.replyq      = rko->rko_replyq.q;
+        rko->rko_replyq.q = NULL; /* Take ownership of replyq */


Keep all the rkcg assignment operations together

Pratyush Ranjan (PratRanj07) · 2026-04-06T17:48:58Z

src/rdkafka.c

+                        if (result_rktpar)
+                                result_rktpar->err =
+                                    RD_KAFKA_RESP_ERR_LEADER_NOT_AVAILABLE;


Shouldn't it be problematic if result_rktpar is not found? Maybe add a debug log or rd_assert in that case to make sure it should always be there?

Pratyush Ranjan (PratRanj07) · 2026-04-06T18:47:26Z

src/rdkafka_cgrp.h

                                                           * reply.
                                                           *   @locality main
                                                           * thread */
+                int64_t commit_sync_id_counter; /**< Global counter for


Suggested change

int64_t commit_sync_id_counter; /**< Global counter for

int64_t commit_sync_request_id_counter; /**< Global counter for

Pratyush Ranjan (PratRanj07) · 2026-04-06T18:49:14Z

src/rdkafka_cgrp.h

+                    *results;                 /**< Partition-to-error mapping.
+                                               *   Populated as broker replies
+                                               *   arrive. */
+                int wait_broker_result_count; /**< Number of broker


Suggested change

int wait_broker_result_count; /**< Number of broker

int brokers_awaiting_result_cnt; /**< Number of broker

or something similar

Pratyush Ranjan (PratRanj07) · 2026-04-06T19:00:33Z

src/rdkafka_share_acknowledgement.c

+                                        rd_kafka_dbg(
+                                            rkshare->rkshare_rk, CGRP,
+                                            "SHAREACK",
+                                            "    Adding ack detail for offsets "


Why this extra space?

Pratyush Ranjan (PratRanj07) · 2026-04-06T19:01:44Z

tests/0176-share_consumer_commit_sync.c

+ * and returns per-partition results.
+ */
+
+#define MAX_MSGS      500


This is not needed

Ojasva Jain (ojasvajain) · 2026-04-07T06:54:14Z

src/rdkafka.c

+        rd_kafka_share_acknowledge_all_if_implicit(rkshare);
+
+        ack_batches = rd_kafka_share_build_ack_details(rk->rk_rkshare);


If multiple app threads call commitSync(), there can be race conditions while building the acks.

Ojasva Jain (ojasvajain) · 2026-04-07T14:36:18Z

src/rdkafka.c

+                rd_kafka_topic_partition_t *result_rktpar;
+
+                rktp = rd_kafka_topic_partition_toppar(rk, batch->rktpar);
+                if (!rktp || !rktp->rktp_leader) {


Should we wrap this condition in unlikely()?

Ojasva Jain (ojasvajain) · 2026-04-07T19:14:07Z

src/rdkafka_broker.h

+                rd_ts_t abs_timeout;            /**< Absolute timeout from
+                                                 *   the commit_sync request. */
+                int64_t commit_sync_request_id; /**< Request ID this


Can we get rid of these fields? They are only getting used while creating the rko to be sent from MT to BT. In that situation, we can use values from cgrp struct instead.

Ojasva Jain (ojasvajain) · 2026-04-07T19:42:18Z

src/rdkafka.c

+         * Step 3: Dispatch pending commit_sync to the replying broker
+         * (highest priority). Acks must always be sent.
+         */
+        if (!reply_rkb->rkb_share_fetch_enqueued &&


Do we need the check on rkb_share_fetch_enqueued? We are setting it to false at the beginning of this function.

Ojasva Jain (ojasvajain) · 2026-04-07T20:25:34Z

src/rdkafka.c

+        rko->rko_u.share_commit_sync_fanout.results     = NULL;
+
+        /* Initialize commit_sync request state */
+        rkcg->rkcg_share.commit_sync_id_counter++;


Instead of having this global counter, can we generate the id using a random no. generator or use abs_timeout value? If yes, we can get rid of commit_sync_id_counter field.

Pranav Rathi (pranavrth) requested a review from a team as a code owner March 30, 2026 13:01

Emanuele Sabellico (emasab) added the consumer label Mar 31, 2026

airlock-confluentinc bot force-pushed the dev_kip-932_ShareAcknowledge-rpc branch from c9e39d3 to 541611f Compare April 1, 2026 14:16

airlock-confluentinc bot force-pushed the dev_kip-932_commit-sync branch from 33f3941 to 0988202 Compare April 1, 2026 14:26

airlock-confluentinc bot force-pushed the dev_kip-932_ShareAcknowledge-rpc branch from 541611f to 469f22b Compare April 2, 2026 01:29

airlock-confluentinc bot force-pushed the dev_kip-932_commit-sync branch from 0988202 to 2d00c1f Compare April 2, 2026 01:44

airlock-confluentinc bot force-pushed the dev_kip-932_ShareAcknowledge-rpc branch from 469f22b to 5e7a5de Compare April 2, 2026 08:26

airlock-confluentinc bot force-pushed the dev_kip-932_commit-sync branch from 43b8005 to 1b69281 Compare April 2, 2026 08:27

Base automatically changed from dev_kip-932_ShareAcknowledge-rpc to dev_kip-932_queues-for-kafka April 6, 2026 07:48

airlock-confluentinc bot force-pushed the dev_kip-932_commit-sync branch from 1b69281 to 551ca4f Compare April 6, 2026 11:53

Pranav Rathi (pranavrth) added 5 commits April 6, 2026 23:58

Style fixes

b57aaea

Moved some tests to local quick tests.

108e46f

Rebased with the base branch

bfa47a3

airlock-confluentinc bot force-pushed the dev_kip-932_commit-sync branch from 551ca4f to bfa47a3 Compare April 6, 2026 18:32

Pranav Rathi (pranavrth) added share consumer and removed consumer labels Apr 6, 2026

Pratyush Ranjan (PratRanj07) requested changes Apr 6, 2026

View reviewed changes

Ojasva Jain (ojasvajain) reviewed Apr 7, 2026

View reviewed changes

	"No pending acknowledgements to commit sync");
	"No pending acknowledgements to commit");

	int64_t commit_sync_id_counter; /**< Global counter for
	int64_t commit_sync_request_id_counter; /**< Global counter for

	int wait_broker_result_count; /**< Number of broker
	int brokers_awaiting_result_cnt; /**< Number of broker

		rd_kafka_share_acknowledge_all_if_implicit(rkshare);

		ack_batches = rd_kafka_share_build_ack_details(rk->rk_rkshare);

Conversation

Pranav Rathi (pranavrth) commented Mar 30, 2026

Uh oh!

confluent-cla-assistant bot commented Mar 30, 2026

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!

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!

4 participants