MockBroker: Implement ShareFetch API (#5302)

* 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]>

Author: 2 people

src/rdkafka_mock.h

@@ -621,6 +621,32 @@ RD_EXPORT void rd_kafka_mock_sharegroup_set_heartbeat_interval(
     rd_kafka_mock_cluster_t *mcluster,
     int heartbeat_interval_ms);
 
+/**
+ * @brief Set the maximum number of times a record can be acquired
+ *        before it is automatically archived (dead-lettered).
+ *
+ * Default is 5.  Set to 0 for unlimited deliveries.
+ *
+ * @param mcluster Mock cluster instance.
+ * @param max_attempts Maximum delivery attempts per record.
+ */
+RD_EXPORT void rd_kafka_mock_sharegroup_set_max_delivery_attempts(
+    rd_kafka_mock_cluster_t *mcluster,
+    int max_attempts);
+
+/**
+ * @brief Set the per-record lock duration in milliseconds.
+ *
+ * When a record is acquired, its lock expires after this duration.
+ * Default is 0, which falls back to the session timeout value.
+ *
+ * @param mcluster Mock cluster instance.
+ * @param lock_duration_ms Lock duration in milliseconds.
+ */
+RD_EXPORT void rd_kafka_mock_sharegroup_set_record_lock_duration(
+    rd_kafka_mock_cluster_t *mcluster,
+    int lock_duration_ms);
+
 /**
  * @brief Set a manual target assignment for a sharegroup.
  *