[KIP-848] Added ConsumerGroupHeartbeat API request

and response with basic cgrp_consumer flow
- Added topic id to topic_partition_t while reading from buffer
- Added new methods and way to add topic_id to topic partition private
- Added new configs group.protocol and group.remote.assignor
- Added ConsumerGroupHeartbeat API Request Contract
- Added ConsumerGroupApi Request without handling different cases
- Working ConsumerGroupHeartbeat API with proper response
- Properly receiving assigned partitions with topic_id from the Response
- Added metadata request as well after parsing the response. Separate it out from to the topic partition branch
- Added metadata response flow with rko
- Updated OffsetFetch to v9
- Removed unrequred fields from ConsumerGroupHeartbeat API to make it work with AK > 3.6
- OffsetFetch working fine. Able to consume.
- Changed subsribed list to use correct field
- Fixed few memory leaks
- Some more memory leak fixes. Added updation to subscribed topics list
- Minor changes
- [KIP-848] Added new configs group.protocol and group.remote.assignor (#4414)
- Added new configs group.protocol and group.remote.assignor
- Removed printfs and Updated hardcoded one topic for metadata request
- Removed some changes related to removed changes in the protocols
- [KIP-848] Added topic id to topic_partition_t while reading from buffer (#4416)
- Updating topic name from metadata response for all the requested topic_ids instead of just 1.
- Style fixes and fixed skip tag issue in buf_*_topic_partition
- Changed variable for next assignment
- Added topic name while reading topic partition buffer
- Changed variable name from assignments to assignment

Author: pranavrth

CONFIGURATION.md

@@ -1,7 +1,7 @@
 # Configuration properties
 ## Global configuration properties
 
-Property                                 | C/P | Range           |       Default | Importance | Description              
+Property                                 | C/P | Range           |       Default | Importance | Description
 -----------------------------------------|-----|-----------------|--------------:|------------| --------------------------
 builtin.features                         |  *  |                 | gzip, snappy, ssl, sasl, regex, lz4, sasl_gssapi, sasl_plain, sasl_scram, plugins, zstd, sasl_oauthbearer, http, oidc | low        | Indicates the builtin features for this build of librdkafka. An application can either query this value or attempt to set it with its list of required features to check for library support. <br>*Type: CSV flags*
 client.id                                |  *  |                 |       rdkafka | low        | Client identifier. <br>*Type: string*
@@ -158,7 +158,7 @@ client.dns.lookup                        |  *  | use_all_dns_ips, resolve_canoni
 
 ## Topic configuration properties
 
-Property                                 | C/P | Range           |       Default | Importance | Description              
+Property                                 | C/P | Range           |       Default | Importance | Description
 -----------------------------------------|-----|-----------------|--------------:|------------| --------------------------
 request.required.acks                    |  P  | -1 .. 1000      |            -1 | high       | This field indicates the number of acknowledgements the leader broker must receive from ISR brokers before responding to the request: *0*=Broker does not send any response/ack to client, *-1* or *all*=Broker will block until message is committed by all in sync replicas (ISRs). If there are less than `min.insync.replicas` (broker configuration) in the ISR set the produce request will fail. <br>*Type: integer*
 acks                                     |  P  | -1 .. 1000      |            -1 | high       | Alias for `request.required.acks`: This field indicates the number of acknowledgements the leader broker must receive from ISR brokers before responding to the request: *0*=Broker does not send any response/ack to client, *-1* or *all*=Broker will block until message is committed by all in sync replicas (ISRs). If there are less than `min.insync.replicas` (broker configuration) in the ISR set the produce request will fail. <br>*Type: integer*

examples/consumer.c

@@ -127,6 +127,20 @@ int main(int argc, char **argv) {
                 return 1;
         }
 
+        if (rd_kafka_conf_set(conf, "group.protocol", "consumer", errstr,
+                              sizeof(errstr)) != RD_KAFKA_CONF_OK) {
+                fprintf(stderr, "%s\n", errstr);
+                rd_kafka_conf_destroy(conf);
+                return 1;
+        }
+
+        //        if (rd_kafka_conf_set(conf, "debug", "all", errstr,
+        //                              sizeof(errstr)) != RD_KAFKA_CONF_OK) {
+        //                fprintf(stderr, "%s\n", errstr);
+        //                rd_kafka_conf_destroy(conf);
+        //                return 1;
+        //        }
+
         /* If there is no previously committed offset for a partition
          * the auto.offset.reset strategy will be used to decide where
          * in the partition to start fetching messages.

src/rdkafka_admin.c

@@ -3939,7 +3939,7 @@ rd_kafka_DeleteRecordsResponse_parse(rd_kafka_op_t *rko_req,
             RD_KAFKA_TOPIC_PARTITION_FIELD_ERR,
             RD_KAFKA_TOPIC_PARTITION_FIELD_END};
         offsets = rd_kafka_buf_read_topic_partitions(
-            reply, rd_false /* don't use topic_id */, 0, fields);
+            reply, rd_false /* don't use topic_id */, rd_true, 0, fields);
         if (!offsets)
                 rd_kafka_buf_parse_fail(reply,
                                         "Failed to parse topic partitions");
@@ -4926,7 +4926,7 @@ rd_kafka_OffsetDeleteResponse_parse(rd_kafka_op_t *rko_req,
             RD_KAFKA_TOPIC_PARTITION_FIELD_ERR,
             RD_KAFKA_TOPIC_PARTITION_FIELD_END};
         partitions = rd_kafka_buf_read_topic_partitions(
-            reply, rd_false /* don't use topic_id */, 16, fields);
+            reply, rd_false /* don't use topic_id */, rd_true, 16, fields);
         if (!partitions) {
                 rd_snprintf(errstr, errstr_size,
                             "Failed to parse OffsetDeleteResponse partitions");
@@ -6961,8 +6961,8 @@ static rd_kafka_resp_err_t rd_kafka_ListConsumerGroupOffsetsRequest(
         require_stable_offsets =
             rd_kafka_confval_get_int(&options->require_stable_offsets);
         rd_kafka_OffsetFetchRequest(
-            rkb, grpoffsets->group_id, grpoffsets->partitions,
-            require_stable_offsets, op_timeout, replyq, resp_cb, opaque);
+            rkb, grpoffsets->group_id, grpoffsets->partitions, rd_false, -1,
+            NULL, require_stable_offsets, op_timeout, replyq, resp_cb, opaque);
         return RD_KAFKA_RESP_ERR_NO_ERROR;
 }
 
@@ -8114,8 +8114,8 @@ rd_kafka_DescribeConsumerGroupsResponse_parse(rd_kafka_op_t *rko_req,
                                     {RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
                                      RD_KAFKA_TOPIC_PARTITION_FIELD_END};
                                 partitions = rd_kafka_buf_read_topic_partitions(
-                                    rkbuf, rd_false /* don't use topic_id */, 0,
-                                    fields);
+                                    rkbuf, rd_false /* don't use topic_id */,
+                                    rd_true, 0, fields);
                                 rd_kafka_buf_destroy(rkbuf);
                                 if (!partitions)
                                         rd_kafka_buf_parse_fail(

src/rdkafka_assignment.c

@@ -543,7 +543,8 @@ static int rd_kafka_assignment_serve_pending(rd_kafka_t *rk) {
                              partitions_to_query->cnt);
 
                 rd_kafka_OffsetFetchRequest(
-                    coord, rk->rk_group_id->str, partitions_to_query,
+                    coord, rk->rk_group_id->str, partitions_to_query, rd_false,
+                    -1, NULL,
                     rk->rk_conf.isolation_level ==
                         RD_KAFKA_READ_COMMITTED /*require_stable_offsets*/,
                     0, /* Timeout */

src/rdkafka_assignor.c

@@ -159,8 +159,8 @@ rd_kafkap_bytes_t *rd_kafka_consumer_protocol_member_metadata_new(
                 rd_kafka_buf_write_topic_partitions(
                     rkbuf, owned_partitions,
                     rd_false /*don't skip invalid offsets*/,
-                    rd_false /*any offset*/, rd_false /* don't use topic_id */,
-                    fields);
+                    rd_false /*any offset*/, rd_false /* use_topic name */,
+                    rd_true, fields);
         }
 
         /* Following data is ignored by consumer version < 2 */

src/rdkafka_buf.c

@@ -38,6 +38,8 @@ void rd_kafka_buf_destroy_final(rd_kafka_buf_t *rkbuf) {
         case RD_KAFKAP_Metadata:
                 if (rkbuf->rkbuf_u.Metadata.topics)
                         rd_list_destroy(rkbuf->rkbuf_u.Metadata.topics);
+                if (rkbuf->rkbuf_u.Metadata.topic_ids)
+                        rd_list_destroy(rkbuf->rkbuf_u.Metadata.topic_ids);
                 if (rkbuf->rkbuf_u.Metadata.reason)
                         rd_free(rkbuf->rkbuf_u.Metadata.reason);
                 if (rkbuf->rkbuf_u.Metadata.rko)

src/rdkafka_buf.h

@@ -375,8 +375,10 @@ struct rd_kafka_buf_s { /* rd_kafka_buf_t */
 
         union {
                 struct {
-                        rd_list_t *topics;     /* Requested topics (char *) */
-                        char *reason;          /* Textual reason */
+                        rd_list_t *topics; /* Requested topics (char *) */
+                        rd_list_t *
+                            topic_ids; /* Requested topic ids rd_kafka_Uuid_t */
+                        char *reason;  /* Textual reason */
                         rd_kafka_op_t *rko;    /* Originating rko with replyq
                                                 * (if any) */
                         rd_bool_t all_topics;  /**< Full/All topics requested */