[KIP-932] Fix intermittent test and add debug logging

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

Author: pranavrth

src/rdkafka.c

@@ -3443,6 +3443,20 @@ static void rd_kafka_share_segregate_acks_by_leader(rd_kafka_t *rk,
                          * is then fully destroyed (freeing its entries). */
                         rd_kafka_share_ack_batch_entry_t *entry;
                         int j;
+                        /*
+                         * TODO KIP-932: Merge with overlapping offsets into
+                         * single entry with single type. For example, if
+                         * existing has an ACCEPT for offsets 0-50 and new
+                         * batch has ACCEPT for offsets 51-100, they can be
+                         * merged into a single ACCEPT for offsets 0-100.
+                         * Currently, we are blindly adding entries from the
+                         * new batch into the existing batch, which means we
+                         * may end up with multiple adjacent or overlapping
+                         * entries of the same type that could have been
+                         * merged. This is not incorrect, but it is less
+                         * efficient in terms of the network bandwidth in the
+                         * RPC call.
+                         */
                         RD_LIST_FOREACH(entry, &batch->entries, j) {
                                 rd_list_add(
                                     &existing->entries,
@@ -4212,6 +4226,8 @@ rd_kafka_error_t *rd_kafka_share_commit_async(rd_kafka_share_t *rkshare) {
         rd_kafka_op_t *rko;
         rd_list_t *ack_batches;
 
+        rd_kafka_dbg(rk, CGRP, "SHARE", "Committing asynchronously");
+
         /* Drain rk_rep for all pending callbacks (non-blocking) */
         rd_kafka_q_serve(rk->rk_rep, RD_POLL_NOWAIT, 0, RD_KAFKA_Q_CB_CALLBACK,
                          rd_kafka_poll_cb, NULL);
@@ -4274,6 +4290,9 @@ rd_kafka_share_commit_sync(rd_kafka_share_t *rkshare,
 
         *partitions = NULL;
 
+        rd_kafka_dbg(rk, CGRP, "SHARE",
+                     "Committing synchronously with timeout %d ms", timeout_ms);
+
         /* Drain rk_rep for all pending callbacks (non-blocking) */
         rd_kafka_q_serve(rk->rk_rep, RD_POLL_NOWAIT, 0, RD_KAFKA_Q_CB_CALLBACK,
                          rd_kafka_poll_cb, NULL);

src/rdkafka_fetcher.c

@@ -2651,6 +2651,11 @@ void rd_kafka_ShareAcknowledgeRequest(rd_kafka_broker_t *rkb,
         /* Topics array with acknowledgement batches */
         of_TopicArrayCnt = rd_kafka_buf_write_arraycnt_pos(rkbuf);
 
+        rd_kafka_dbg(rkb->rkb_rk, FETCH, "SHAREACK",
+                     "Building ShareAcknowledge request with %d ack toppars "
+                     "and %d total ack entries",
+                     ack_details_cnt, total_ack_entries);
+
         if (ack_details) {
                 rd_kafka_Uuid_t *topic_id_last = NULL;
                 rd_kafka_share_ack_batches_t *batches;
@@ -2690,11 +2695,25 @@ void rd_kafka_ShareAcknowledgeRequest(rd_kafka_broker_t *rkb,
                         rd_kafka_buf_write_i32(rkbuf,
                                                batches->rktpar->partition);
 
+                        rd_kafka_dbg(rkb->rkb_rk, FETCH, "SHAREACK",
+                                     "Adding ack for topic %s [%" PRId32
+                                     "] with %d entries",
+                                     batches->rktpar->topic,
+                                     batches->rktpar->partition,
+                                     rd_list_cnt(&batches->entries));
+
                         /* Write acknowledgement batches */
                         entries_cnt = rd_list_cnt(&batches->entries);
                         rd_kafka_buf_write_arraycnt(rkbuf, entries_cnt);
 
                         RD_LIST_FOREACH(entry, &batches->entries, m) {
+                                rd_kafka_dbg(
+                                    rkb->rkb_rk, FETCH, "SHAREACK",
+                                    "Adding ack entry with start offset "
+                                    "%" PRId64 ", end offset %" PRId64
+                                    ", type %d",
+                                    entry->start_offset, entry->end_offset,
+                                    entry->types[0]);
                                 /* FirstOffset */
                                 rd_kafka_buf_write_i64(rkbuf,
                                                        entry->start_offset);

src/rdkafka_share_acknowledgement.c

@@ -372,6 +372,13 @@ rd_list_t *rd_kafka_share_build_ack_details(rd_kafka_share_t *rkshare) {
                                                         ->response_leader_epoch,
                                                     0);
                                         }
+                                        rd_kafka_dbg(
+                                            rkshare->rkshare_rk, CGRP,
+                                            "SHAREACK",
+                                            "    Adding ack detail for offsets "
+                                            "[%" PRId64 " - %" PRId64
+                                            "] with type %d",
+                                            run_start, run_end, run_type);
                                         /* Collated: 1 type for entire
                                          * range */
                                         rd_list_add(