MockBroker: Implement ShareFetch API by k-raina · Pull Request #5302 · confluentinc/librdkafka

Kaushik Raina (k-raina) · 2026-02-06T13:00:59Z

Summary

Implemented KIP‑932 ShareFetch in the mock broker, including session management, partition metadata (SPSO/SPEO), inflight record state, acquisition logic, and response construction.
Wired ShareFetch/ShareGroupHeartbeat ApiVersions, fixed client ShareFetch request versioning, and ensured SGHB sends subscription topics.
Added a comprehensive ShareFetch mock‑broker test suite (share consumer) with positive/negative scenarios and RTT‑expiry coverage.

ShareFetch Workflow

[1] Parse Request
  - Read GroupId, MemberId, SessionEpoch
  - Read Wait/Min/Max/MaxRecords/BatchSize
  - Read Topics -> Partitions -> AckBatches/AckTypes
  - Skip partition/topic/top-level tags

        |
        v

[2] Resolve ShareGroup
  - Lookup by GroupId
  - Create new group if missing
  - Start session timer (if new)

        |
        v

[3] Session Management
  - Find session by member/session id
  - Validate SessionEpoch
  - Update session partitions if changed
  - Update last-activity timestamp

        |
        v

[4] Partition Metadata + Acquisition
  - Validate topic + partition exist
  - Get/create SharePartitionMeta (SPSO/SPEO)
  - Prune archived inflight records
  - Acquire available offsets with caps

        |
        v

[5] Build ShareFetch Response
  - Write ThrottleTime, Error, LockTimeout
  - For each topic/partition:
      * Errors + CurrentLeader
      * Records (msgset)
      * AcquiredRecords
      * Tags
  - Write NodeEndpoints + top-level tags

        |
        v

[6] Client Parse + Delivery
  - Parse response + records
  - Store acquired records for ack
  - Deliver messages to app

High level Classe and Data structures

MockCluster
  |
  +-- shareGroups : TAILQ<MockShareGroup>
         |
         +-- id : String
         +-- groupEpoch : int
         +-- sessionTimeoutMs : int
         |
         +-- fetchSessions : TAILQ<ShareFetchSession>
         |       |
         |       +-- memberId : String
         |       +-- sessionId : int
         |       +-- sessionEpoch : int
         |       +-- lastActivityTs : long
         |       +-- partitions : List<TopicPartition>
         |
         +-- partitions : TAILQ<SharePartitionMeta>
                 |
                 +-- topicId : UUID
                 +-- partition : int
                 +-- spso : long
                 +-- speo : long
                 +-- inflight : TAILQ<RecordState>
                         |
                         +-- offset : long
                         +-- ownerMemberId : String
                         +-- lockExpiryTs : long
                         +-- deliveryCount : int
                         +-- state : RecordStateState
                                   { AVAILABLE | ACQUIRED | ARCHIVED }

confluent-cla-assistant · 2026-02-06T13: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.}

Ankith L (Ankith-Confluent) · 2026-02-09T09:44:04Z

src/rdkafka_fetcher.c


        ApiVersion = rd_kafka_broker_ApiVersion_supported(rkb, RD_KAFKAP_ShareFetch,
-                                                          1, 1, NULL);
+                                                          0, 0, NULL);


Java version of Share Fetch supports version (1, 1) and above, so I think this should also be the same

Ankith L (Ankith-Confluent) · 2026-02-09T09:56:01Z

src/rdkafka_mock.c

        rd_kafka_mock_cgrp_classic_t *mcgrp_classic;
        rd_kafka_mock_cgrp_consumer_t *mcgrp_consumer;
        rd_kafka_mock_sharegroup_t *mshgrp;
+        rd_kafka_mock_sgrp_t *msgrp;


This is actually duplicate.
rd_kafka_mock_sharegroup_t already exists

Ankith L (Ankith-Confluent) · 2026-02-09T09:58:30Z

src/rdkafka_mock.c

        while ((mshgrp = TAILQ_FIRST(&mcluster->sharegrps)))
                rd_kafka_mock_sharegroup_destroy(mshgrp);

+        while ((msgrp = TAILQ_FIRST(&mcluster->sgrps_share)))


same regarding this, this is also duplicate.

Ankith L (Ankith-Confluent) · 2026-02-09T09:58:50Z

src/rdkafka_mock.c

        TAILQ_INIT(&mcluster->cgrps_consumer);

        rd_kafka_mock_sharegrps_init(mcluster);
+        TAILQ_INIT(&mcluster->sgrps_share);


The init is already present
rd_kafka_mock_sharegrps_init(mcluster);

Ankith L (Ankith-Confluent) · 2026-02-09T10:34:16Z

src/rdkafka_mock_cgrp.c

+ * @locks mcluster->lock MUST be held.
+ */
+rd_kafka_mock_sgrp_t *
+rd_kafka_mock_sgrp_find(rd_kafka_mock_cluster_t *mcluster,


This is already present in src/rdkafka_mock_sharegrp.c

Ankith L (Ankith-Confluent) · 2026-02-09T10:47:45Z

src/rdkafka_mock_int.h

+rd_kafka_mock_sgrp_find(rd_kafka_mock_cluster_t *mcluster,
+                        const rd_kafkap_str_t *GroupId);
+rd_kafka_mock_sgrp_t *
+rd_kafka_mock_sgrp_get(rd_kafka_mock_cluster_t *mcluster,


find, get, and destroy are duplicates

Ankith L (Ankith-Confluent) · 2026-02-09T10:58:55Z

src/rdkafka_mock_handlers.c

        }

-        rd_kafka_buf_skip_tags(rkbuf);
+        /* ForgottenToppars */


I didnt understand why this code block is present in the metadata_handler.
Shouldnt this be part of share fetch handler

Ankith L (Ankith-Confluent) · 2026-02-09T11:30:31Z

src/rdkafka_mock_handlers.c

+        TAILQ_FOREACH_SAFE(state, &pmeta->inflight, link, tmp) {
+                if (state->state != RD_KAFKA_MOCK_SGRP_RECORD_ARCHIVED)
+                        continue;
+                if (state->offset <= pmeta->speo)


The messages will not be pruned in every scenario, because the offset can never be more than speo.
So it should be state->offset >= pmeta->spso instead.

Ankith L (Ankith-Confluent) · 2026-02-09T11:43:55Z

src/rdkafka_mock_handlers.c

+                                        rd_kafka_buf_write_kbytes(
+                                            resp, mset ? &mset->bytes : NULL);
+                                        /* Response: AcquiredRecords */
+                                        rd_kafka_buf_write_arraycnt(resp, 0);


Is there a reason for it to be 0?

Yes, we have kept acknolegement workflow seperate

Ankith L (Ankith-Confluent) · 2026-02-09T11:46:38Z

src/rdkafka_mock_handlers.c

+}
+
+static const rd_kafka_mock_msgset_t *
+rd_kafka_mock_sgrp_first_acquired_msgset(


why are we sending only the first message set?
Shouldnt we send all the sets in the acquired offset range

Ankith L (Ankith-Confluent) · 2026-02-10T08:34:06Z

src/rdkafka_mock_cgrp.c

        rd_free(mcgrp);
 }

+/**


can we shift these functions to rdkafka_mock_sharegrp.c

Ankith L (Ankith-Confluent) · 2026-02-10T08:43:57Z

src/rdkafka_mock_cgrp.c

+ * @brief Destroy share-partition metadata.
+ */
+static void
+rd_kafka_mock_sgrp_partmeta_destroy(rd_kafka_mock_sgrp_partmeta_t *pmeta) {


This function is not being used anywhere
what is the use of this?

Actually sharegroup_destroy() handles it, so this is not required I guess

Ankith L (Ankith-Confluent) · 2026-02-10T08:48:01Z

src/rdkafka_mock_cgrp.c

+                                                        session->member_id);
+
+                TAILQ_REMOVE(&mshgrp->fetch_sessions, session, link);
+                mshgrp->fetch_session_cnt--;


I think this requires lock to modify share group related data?

Ankith L (Ankith-Confluent) · 2026-02-10T08:52:07Z

src/rdkafka_mock_handlers.c

+    const rd_kafka_mock_partition_t *mpart,
+    const rd_kafkap_str_t *member_id,
+    rd_ts_t now) {
+        const rd_kafka_mock_msgset_t *msgsets[256];


is 256 intentional?

Ankith L (Ankith-Confluent)

Hey Kaushik Raina (@k-raina)
Thanks for the PR!
LGTM!

- Introduced functions to set session timeout and heartbeat interval for sharegroups. - Implemented manual target assignment for sharegroup members. - Enhanced connection handling to clear states for closed connections.

…gnment handling

…to kraina-mock-broker-share-fetch

* Mock Broker : Implement share fetch * ShareGroupHeartBeat MVP * Add sharegroup session management and manual assignment features - Introduced functions to set session timeout and heartbeat interval for sharegroups. - Implemented manual target assignment for sharegroup members. - Enhanced connection handling to clear states for closed connections. * Add function to retrieve member IDs from a sharegroup and update assignment handling * Mock Broker : Implement share fetch * Fix SGHB return and fetcher api version * Add test for share fetch API * Fix session workflow * Update API version * Add handing for ForgottenTopicsData * Add lock expiry logic * Remove duplicates with sghb * Address feedbacks * Implement max delivery attempts and record lock * Add member validation for share fetch * Remove tmp produce code * Implement feedback * Minor * Add test to CMakeList * Fix minor test failure --------- Co-authored-by: Ankith-Confluent <[email protected]>

Mock Broker : Implement share fetch

1b28876

Kaushik Raina (k-raina) requested a review from a team as a code owner February 6, 2026 13:01

Ankith L (Ankith-Confluent) reviewed Feb 9, 2026

View reviewed changes

Kaushik Raina (k-raina) requested a review from Ankith L (Ankith-Confluent) February 10, 2026 06:25

Ankith L (Ankith-Confluent) reviewed Feb 10, 2026

View reviewed changes

Kaushik Raina (k-raina) requested a review from Ankith L (Ankith-Confluent) February 11, 2026 10:47

Ankith L (Ankith-Confluent) reviewed Feb 12, 2026

View reviewed changes

Ankith L (Ankith-Confluent) approved these changes Feb 12, 2026

View reviewed changes

Base automatically changed from anl_MB_SGHB to dev_kip-932_queues-for-kafka February 13, 2026 06:32

Ankith L (Ankith-Confluent) and others added 17 commits February 13, 2026 16:45

ShareGroupHeartBeat MVP

d5521e4

Add sharegroup session management and manual assignment features

ddab80e

- Introduced functions to set session timeout and heartbeat interval for sharegroups. - Implemented manual target assignment for sharegroup members. - Enhanced connection handling to clear states for closed connections.

Add function to retrieve member IDs from a sharegroup and update assi…

26a66e5

…gnment handling

Mock Broker : Implement share fetch

b1a6f0c

Fix SGHB return and fetcher api version

52998d7

Add test for share fetch API

88699bc

Fix session workflow

0723bc9

Update API version

80a42f1

Add handing for ForgottenTopicsData

fce9a47

Add lock expiry logic

1045dc8

Remove duplicates with sghb

4121f89

Address feedbacks

83e7a7a

Implement max delivery attempts and record lock

9b95db5

Add member validation for share fetch

c98ffa3

Remove tmp produce code

1508077

Implement feedback

97f9546

Minor

01874dd

airlock-confluentinc bot force-pushed the kraina-mock-broker-share-fetch branch from 6f6aba9 to 01874dd Compare February 13, 2026 11:17

Kaushik Raina (k-raina) added 2 commits February 13, 2026 17:58

Merge remote-tracking branch 'origin/dev_kip-932_queues-for-kafka' in…

17e9447

…to kraina-mock-broker-share-fetch

Add test to CMakeList

fc71c63

Fix minor test failure

a6c56c6

Ankith L (Ankith-Confluent) self-requested a review February 15, 2026 08:40

Ankith L (Ankith-Confluent) merged commit cac0c73 into dev_kip-932_queues-for-kafka Feb 15, 2026
1 of 2 checks passed

Ankith L (Ankith-Confluent) deleted the kraina-mock-broker-share-fetch branch February 15, 2026 08:44

Conversation

Kaushik Raina (k-raina) commented Feb 6, 2026

Uh oh!

confluent-cla-assistant bot commented Feb 6, 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!

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) left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants