requested changes

Author: PratRanj07

src/rdkafka.c

@@ -3223,6 +3223,13 @@ rd_kafka_op_res_t rd_kafka_share_fetch_reply_op(rd_kafka_t *rk,
                      records_fetched, should_leave,
                      reply_rkb ? rd_kafka_broker_name(reply_rkb) : "none");
 
+        /*
+         * Step 1: Dispatch acknowledgement callbacks.
+         * Per-partition errors were set by the broker thread.
+         */
+        rd_kafka_share_dispatch_ack_callbacks(
+            rk, rko_orig->rko_u.share_fetch.ack_details);
+
         reply_rkb->rkb_share_fetch_enqueued = rd_false;
 
         if (should_fetch)
@@ -3234,15 +3241,15 @@ rd_kafka_op_res_t rd_kafka_share_fetch_reply_op(rd_kafka_t *rk,
         }
 
         /*
-         * Step 1: If records were fetched and broker is not terminating,
+         * Step 2: If records were fetched and broker is not terminating,
          * reset the global fetch guard so the next FANOUT can select
          * a new fetch broker.
          */
         if (records_fetched && rko_orig->rko_err != RD_KAFKA_RESP_ERR__DESTROY)
                 rkcg->rkcg_share.share_fetch_more_records = rd_false;
 
         /*
-         * Step 2: If the consumer has been marked for termination,
+         * Step 3: If the consumer has been marked for termination,
          * enqueue a session leave op on the replying broker thread and return
          */
         if (rkcg->rkcg_flags & RD_KAFKA_CGRP_F_TERMINATE) {
@@ -3252,7 +3259,7 @@ rd_kafka_op_res_t rd_kafka_share_fetch_reply_op(rd_kafka_t *rk,
         }
 
         /*
-         * Step 3: Handle commit_sync reply if this op belongs to
+         * Step 4: Handle commit_sync reply if this op belongs to
          * the current commit_sync request.
          */
         if (rko_orig->rko_u.share_fetch.commit_sync_request_id != 0 &&
@@ -3307,13 +3314,6 @@ rd_kafka_op_res_t rd_kafka_share_fetch_reply_op(rd_kafka_t *rk,
                 rd_kafka_share_commit_sync_maybe_complete(rk, rkcg);
         }
 
-        /*
-         * Step 4: Dispatch acknowledgement callbacks.
-         * Per-partition errors were set by the broker thread.
-         */
-        rd_kafka_share_dispatch_ack_callbacks(
-            rk, rko_orig->rko_u.share_fetch.ack_details);
-
         /*
          * Step 5: Dispatch pending commit_sync to the replying broker
          * (highest priority). Acks must always be sent.
@@ -5725,7 +5725,7 @@ rd_kafka_op_res_t rd_kafka_poll_cb(rd_kafka_t *rk,
                                             rko->rko_u.offset_commit.opaque);
                 break;
 
-        case RD_KAFKA_OP_SHARE_ACK_COMMIT:
+        case RD_KAFKA_OP_SHARE_ACK_COMMIT_CB:
                 if (!rko->rko_u.share_ack_commit.cb)
                         return RD_KAFKA_OP_RES_PASS; /* Dont handle here */
                 rko->rko_u.share_ack_commit.cb(

src/rdkafka_op.c

@@ -131,7 +131,7 @@ const char *rd_kafka_op2str(rd_kafka_op_type_t type) {
             [RD_KAFKA_OP_SHARE_SESSION_PARTITION_REMOVE] =
                 "REPLY:SHARE_SESSION_PARTITION_REMOVE",
             [RD_KAFKA_OP_SHARE_FETCH_RESPONSE] = "REPLY:SHARE_FETCH_RESPONSE",
-            [RD_KAFKA_OP_SHARE_ACK_COMMIT]     = "REPLY:SHARE_ACK_COMMIT",
+            [RD_KAFKA_OP_SHARE_ACK_COMMIT_CB]  = "REPLY:SHARE_ACK_COMMIT_CB",
             [RD_KAFKA_OP_SHARE_COMMIT_ASYNC_FANOUT] =
                 "REPLY:SHARE_COMMIT_ASYNC_FANOUT",
             [RD_KAFKA_OP_SHARE_COMMIT_SYNC_FANOUT] =
@@ -309,7 +309,7 @@ rd_kafka_op_t *rd_kafka_op_new0(const char *source, rd_kafka_op_type_t type) {
             [RD_KAFKA_OP_SHARE_SESSION_PARTITION_REMOVE] = _RD_KAFKA_OP_EMPTY,
             [RD_KAFKA_OP_SHARE_FETCH_RESPONSE] =
                 sizeof(rko->rko_u.share_fetch_response),
-            [RD_KAFKA_OP_SHARE_ACK_COMMIT] =
+            [RD_KAFKA_OP_SHARE_ACK_COMMIT_CB] =
                 sizeof(rko->rko_u.share_ack_commit),
             [RD_KAFKA_OP_SHARE_COMMIT_ASYNC_FANOUT] =
                 sizeof(rko->rko_u.share_commit_async_fanout),
@@ -584,7 +584,7 @@ void rd_kafka_op_destroy(rd_kafka_op_t *rko) {
                 break;
         }
 
-        case RD_KAFKA_OP_SHARE_ACK_COMMIT:
+        case RD_KAFKA_OP_SHARE_ACK_COMMIT_CB:
                 RD_IF_FREE(rko->rko_u.share_ack_commit.partitions,
                            rd_kafka_share_partition_offsets_list_destroy);
                 break;

src/rdkafka_op.h

@@ -212,7 +212,7 @@ typedef enum {
         RD_KAFKA_OP_SHARE_FETCH_RESPONSE, /**< Share fetch response containing
                                            *   all messages and partition acks
                                            *   from a single broker response. */
-        RD_KAFKA_OP_SHARE_ACK_COMMIT,     /**< Share acknowledgement callback
+        RD_KAFKA_OP_SHARE_ACK_COMMIT_CB,  /**< Share acknowledgement callback
                                            *   reply: main -> app */
 
         RD_KAFKA_OP__END

src/rdkafka_share_acknowledgement.c

@@ -792,7 +792,7 @@ rd_kafka_share_acknowledge_offset(rd_kafka_share_t *rkshare,
  * @param partition Partition id.
  * @param offsets_cnt Number of offsets to allocate space for.
  */
-rd_kafka_share_partition_offsets_t *
+static rd_kafka_share_partition_offsets_t *
 rd_kafka_share_partition_offsets_new(rd_kafka_Uuid_t topic_id,
                                      const char *topic,
                                      int32_t partition,
@@ -816,7 +816,7 @@ rd_kafka_share_partition_offsets_new(rd_kafka_Uuid_t topic_id,
  *
  * @param elem Element to destroy.
  */
-void rd_kafka_share_partition_offsets_destroy(
+static void rd_kafka_share_partition_offsets_destroy(
     rd_kafka_share_partition_offsets_t *elem) {
         if (!elem)
                 return;
@@ -832,7 +832,7 @@ void rd_kafka_share_partition_offsets_destroy(
  *          Caller must destroy with
  *          rd_kafka_share_partition_offsets_list_destroy().
  */
-rd_kafka_share_partition_offsets_list_t *
+static rd_kafka_share_partition_offsets_list_t *
 rd_kafka_share_partition_offsets_list_new(int capacity) {
         rd_kafka_share_partition_offsets_list_t *list;
         size_t size;
@@ -897,9 +897,10 @@ rd_kafka_share_build_partition_offsets_list(
 }
 
 
-void rd_kafka_share_enqueue_ack_callback(rd_kafka_t *rk,
-                                         rd_kafka_share_ack_batches_t *batches,
-                                         rd_kafka_resp_err_t err) {
+void rd_kafka_share_enqueue_ack_commit_cb_op(
+    rd_kafka_t *rk,
+    rd_kafka_share_ack_batches_t *batches,
+    rd_kafka_resp_err_t err) {
         rd_kafka_op_t *cb_rko;
         rd_kafka_share_partition_offsets_list_t *partitions;
 
@@ -910,7 +911,7 @@ void rd_kafka_share_enqueue_ack_callback(rd_kafka_t *rk,
         if (!partitions)
                 return;
 
-        cb_rko          = rd_kafka_op_new(RD_KAFKA_OP_SHARE_ACK_COMMIT);
+        cb_rko          = rd_kafka_op_new(RD_KAFKA_OP_SHARE_ACK_COMMIT_CB);
         cb_rko->rko_err = err;
         cb_rko->rko_u.share_ack_commit.partitions = partitions;
         cb_rko->rko_u.share_ack_commit.cb =
@@ -984,9 +985,7 @@ void rd_kafka_share_dispatch_ack_callbacks(rd_kafka_t *rk,
 
         /* Use per-partition error from each batch */
         RD_LIST_FOREACH(ack_batch, ack_details, k) {
-                rd_kafka_share_enqueue_ack_callback(
-                    rk, ack_batch,
-                    ack_batch->rktpar ? ack_batch->rktpar->err
-                                      : RD_KAFKA_RESP_ERR_NO_ERROR);
+                rd_kafka_share_enqueue_ack_commit_cb_op(rk, ack_batch,
+                                                        ack_batch->rktpar->err);
         }
 }

src/rdkafka_share_acknowledgement.h

@@ -28,10 +28,6 @@
 #ifndef _RDKAFKA_SHARE_ACKNOWLEDGEMENT_H_
 #define _RDKAFKA_SHARE_ACKNOWLEDGEMENT_H_
 
-#include "rdlist.h"
-#include "rdkafka.h"
-
-/* Forward declarations */
 typedef struct rd_kafka_op_s rd_kafka_op_t;
 
 typedef enum rd_kafka_internal_ShareAcknowledgement_type_s {
@@ -250,39 +246,6 @@ struct rd_kafka_share_partition_offsets_list_s {
                                                           offsets */
 };
 
-/**
- * @brief Allocate and initialize a partition offsets element.
- *
- * @param topic_id Topic UUID.
- * @param topic Topic name (will be duplicated).
- * @param partition Partition id.
- * @param offsets_cnt Number of offsets to allocate space for.
- * @returns Newly allocated partition offsets element.
- */
-rd_kafka_share_partition_offsets_t *
-rd_kafka_share_partition_offsets_new(rd_kafka_Uuid_t topic_id,
-                                     const char *topic,
-                                     int32_t partition,
-                                     int offsets_cnt);
-
-/**
- * @brief Destroy a partition offsets element.
- *
- * @param elem Element to destroy.
- */
-void rd_kafka_share_partition_offsets_destroy(
-    rd_kafka_share_partition_offsets_t *elem);
-
-/**
- * @brief Allocate and initialize a partition offsets list.
- *
- * @param capacity Initial capacity for elements.
- * @returns Newly allocated list, or NULL if capacity is 0.
- *          Caller must destroy with
- *          rd_kafka_share_partition_offsets_list_destroy().
- */
-rd_kafka_share_partition_offsets_list_t *
-rd_kafka_share_partition_offsets_list_new(int capacity);
 
 /**
  * @brief Destroy a partition offsets list.
@@ -320,21 +283,28 @@ rd_kafka_share_build_partition_offsets_list(
  * @param batches The ack batches for this partition (contains offsets).
  * @param err Error code to report in callback.
  */
-void rd_kafka_share_enqueue_ack_callback(rd_kafka_t *rk,
-                                         rd_kafka_share_ack_batches_t *batches,
-                                         rd_kafka_resp_err_t err);
+void rd_kafka_share_enqueue_ack_commit_cb_op(
+    rd_kafka_t *rk,
+    rd_kafka_share_ack_batches_t *batches,
+    rd_kafka_resp_err_t err);
 
 
 /**
- * @brief Dispatch ack callbacks for all partitions in ack_details.
+ * @brief Enqueue acknowledgement callbacks to application for each partition.
+ *
+ * Iterates through each partition in ack_details and enqueues one callback
+ * operation (RD_KAFKA_OP_SHARE_ACK_COMMIT_CB) per partition to the
+ * application's reply queue. Each operation contains:
+ * - The partition's acknowledged offsets
+ * - Per-partition error code from batch->rktpar->err
  *
- * If err is set (top-level error), all partitions receive the same error.
- * Otherwise, per-partition results from ack_results are used.
+ * The application's share_acknowledgement_commit_cb is invoked once per
+ * partition when the app calls rd_kafka_consumer_poll() or
+ * rd_kafka_queue_poll().
  *
  * @param rk Kafka handle.
- * @param ack_details List of ack batches.
- * @param ack_results Per-partition results (may be NULL).
- * @param err Top-level error code.
+ * @param ack_details List of rd_kafka_share_ack_batches_t* with acknowledgement
+ *                    results and per-partition error codes.
  */
 void rd_kafka_share_dispatch_ack_callbacks(rd_kafka_t *rk,
                                            rd_list_t *ack_details);