[KIP-848] topic_id in topic_partition_internal, some new error codes (#4404)

* KIP-848 new error codes (partial)

* Add topic id to topic_partition_private

* Rename rd_kafka_uuid_t to rd_kafka_Uuid_t

* [KIP-848] Added new configs group.protocol and group.remote.assignor (#4414)

Added new configs group.protocol and group.remote.assignor

* [KIP-848] Added topic id to topic_partition_t while reading from buffer (#4416)

* Address comments

* Address comment

* Upgrade vcpkg

---------

Co-authored-by: Pranav Rathi <[email protected]>

Author: emasab

CONFIGURATION.md

@@ -109,7 +109,7 @@ group.instance.id                        |  C  |                 |
 partition.assignment.strategy            |  C  |                 | range,roundrobin | medium     | The name of one or more partition assignment strategies. The elected group leader will use a strategy supported by all members of the group to assign partitions to group members. If there is more than one eligible strategy, preference is determined by the order of this list (strategies earlier in the list have higher priority). Cooperative and non-cooperative (eager) strategies must not be mixed. Available strategies: range, roundrobin, cooperative-sticky. <br>*Type: string*
 session.timeout.ms                       |  C  | 1 .. 3600000    |         45000 | high       | Client group session and failure detection timeout. The consumer sends periodic heartbeats (heartbeat.interval.ms) to indicate its liveness to the broker. If no hearts are received by the broker for a group member within the session timeout, the broker will remove the consumer from the group and trigger a rebalance. The allowed range is configured with the **broker** configuration properties `group.min.session.timeout.ms` and `group.max.session.timeout.ms`. Also see `max.poll.interval.ms`. <br>*Type: integer*
 heartbeat.interval.ms                    |  C  | 1 .. 3600000    |          3000 | low        | Group session keepalive heartbeat interval. <br>*Type: integer*
-group.protocol.type                      |  C  |                 |      consumer | low        | Group protocol type. NOTE: Currently, the only supported group protocol type is `consumer`. <br>*Type: string*
+group.protocol.type                      |  C  |                 |      consumer | low        | Group protocol type for the `generic` group protocol. NOTE: Currently, the only supported group protocol type is `consumer`. <br>*Type: string*
 coordinator.query.interval.ms            |  C  | 1 .. 3600000    |        600000 | low        | How often to query for the current client group coordinator. If the currently assigned coordinator is down the configured query interval will be divided by ten to more quickly recover in case of coordinator reassignment. <br>*Type: integer*
 max.poll.interval.ms                     |  C  | 1 .. 86400000   |        300000 | high       | Maximum allowed time between calls to consume messages (e.g., rd_kafka_consumer_poll()) for high-level consumers. If this interval is exceeded the consumer is considered failed and the group will rebalance in order to reassign the partitions to another consumer group member. Warning: Offset commits may be not possible at this point. Note: It is recommended to set `enable.auto.offset.store=false` for long-time processing applications and then explicitly store offsets (using offsets_store()) *after* message processing, to make sure offsets are not auto-committed prior to processing has finished. The interval is checked two times per second. See KIP-62 for more information. <br>*Type: integer*
 enable.auto.commit                       |  C  | true, false     |          true | high       | Automatically and periodically commit offsets in the background. Note: setting this to false does not prevent the consumer from fetching previously committed start offsets. To circumvent this behaviour set specific start offsets per partition in the call to assign(). <br>*Type: boolean*

src/rdkafka.c

@@ -700,6 +700,10 @@ static const struct rd_kafka_err_desc rd_kafka_err_descs[] = {
     _ERR_DESC(RD_KAFKA_RESP_ERR_PRINCIPAL_DESERIALIZATION_FAILURE,
               "Broker: Request principal deserialization failed during "
               "forwarding"),
+    _ERR_DESC(RD_KAFKA_RESP_ERR_FENCED_MEMBER_EPOCH,
+              "Broker: The member epoch is fenced by the group coordinator"),
+    _ERR_DESC(RD_KAFKA_RESP_ERR_STALE_MEMBER_EPOCH,
+              "Broker: The member epoch is stale"),
     _ERR_DESC(RD_KAFKA_RESP_ERR__END, NULL)};
 
 
@@ -2436,8 +2440,9 @@ rd_kafka_t *rd_kafka_new(rd_kafka_type_t type,
 
                 if (RD_KAFKAP_STR_LEN(rk->rk_group_id) > 0) {
                         /* Create consumer group handle */
-                        rk->rk_cgrp = rd_kafka_cgrp_new(rk, rk->rk_group_id,
-                                                        rk->rk_client_id);
+                        rk->rk_cgrp = rd_kafka_cgrp_new(
+                            rk, rk->rk_conf.group_protocol, rk->rk_group_id,
+                            rk->rk_client_id);
                         rk->rk_consumer.q =
                             rd_kafka_q_keep(rk->rk_cgrp->rkcg_q);
                 } else {

src/rdkafka.h

@@ -631,7 +631,10 @@ typedef enum {
         RD_KAFKA_RESP_ERR_FEATURE_UPDATE_FAILED = 96,
         /** Request principal deserialization failed during forwarding */
         RD_KAFKA_RESP_ERR_PRINCIPAL_DESERIALIZATION_FAILURE = 97,
-
+        /** The member epoch is fenced by the group coordinator */
+        RD_KAFKA_RESP_ERR_FENCED_MEMBER_EPOCH = 110,
+        /** The member epoch is stale */
+        RD_KAFKA_RESP_ERR_STALE_MEMBER_EPOCH = 113,
         RD_KAFKA_RESP_ERR_END_ALL,
 } rd_kafka_resp_err_t;
 
@@ -4546,6 +4549,20 @@ rd_kafka_consumer_group_metadata_new_with_genid(const char *group_id,
                                                 const char *group_instance_id);
 
 
+/**
+ * @brief Get member id of a group metadata.
+ *
+ * @param group_metadata The group metadata
+ *
+ * @returns The member id contained in the passed \p group_metadata.
+ *
+ * @remark The returned pointer has the same lifetime as \p group_metadata.
+ */
+RD_EXPORT
+const char *rd_kafka_consumer_group_metadata_member_id(
+    const rd_kafka_consumer_group_metadata_t *group_metadata);
+
+
 /**
  * @brief Frees the consumer group metadata object as returned by
  *        rd_kafka_consumer_group_metadata().

src/rdkafka_admin.c

@@ -3938,7 +3938,8 @@ rd_kafka_DeleteRecordsResponse_parse(rd_kafka_op_t *rko_req,
             RD_KAFKA_TOPIC_PARTITION_FIELD_OFFSET,
             RD_KAFKA_TOPIC_PARTITION_FIELD_ERR,
             RD_KAFKA_TOPIC_PARTITION_FIELD_END};
-        offsets = rd_kafka_buf_read_topic_partitions(reply, 0, fields);
+        offsets = rd_kafka_buf_read_topic_partitions(
+            reply, rd_false /* don't use topic_id */, 0, fields);
         if (!offsets)
                 rd_kafka_buf_parse_fail(reply,
                                         "Failed to parse topic partitions");
@@ -4924,7 +4925,8 @@ rd_kafka_OffsetDeleteResponse_parse(rd_kafka_op_t *rko_req,
             RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
             RD_KAFKA_TOPIC_PARTITION_FIELD_ERR,
             RD_KAFKA_TOPIC_PARTITION_FIELD_END};
-        partitions = rd_kafka_buf_read_topic_partitions(reply, 16, fields);
+        partitions = rd_kafka_buf_read_topic_partitions(
+            reply, rd_false /* don't use topic_id */, 16, fields);
         if (!partitions) {
                 rd_snprintf(errstr, errstr_size,
                             "Failed to parse OffsetDeleteResponse partitions");
@@ -8112,7 +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, 0, fields);
+                                    rkbuf, rd_false /* don't use topic_id */, 0,
+                                    fields);
                                 rd_kafka_buf_destroy(rkbuf);
                                 if (!partitions)
                                         rd_kafka_buf_parse_fail(

src/rdkafka_assignor.c

@@ -159,7 +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*/, fields);
+                    rd_false /*any offset*/, rd_false /* don't use topic_id */,
+                    fields);
         }
 
         /* Following data is ignored by consumer version < 2 */

src/rdkafka_cgrp.c

@@ -361,6 +361,8 @@ void rd_kafka_cgrp_destroy_final(rd_kafka_cgrp_t *rkcg) {
         rd_kafka_cgrp_set_member_id(rkcg, NULL);
         if (rkcg->rkcg_group_instance_id)
                 rd_kafkap_str_destroy(rkcg->rkcg_group_instance_id);
+        if (rkcg->rkcg_group_remote_assignor)
+                rd_kafkap_str_destroy(rkcg->rkcg_group_remote_assignor);
 
         rd_kafka_q_destroy_owner(rkcg->rkcg_q);
         rd_kafka_q_destroy_owner(rkcg->rkcg_ops);
@@ -398,18 +400,20 @@ rd_kafka_cgrp_update_session_timeout(rd_kafka_cgrp_t *rkcg, rd_bool_t reset) {
 
 
 rd_kafka_cgrp_t *rd_kafka_cgrp_new(rd_kafka_t *rk,
+                                   rd_kafka_group_protocol_t group_protocol,
                                    const rd_kafkap_str_t *group_id,
                                    const rd_kafkap_str_t *client_id) {
         rd_kafka_cgrp_t *rkcg;
 
         rkcg = rd_calloc(1, sizeof(*rkcg));
 
-        rkcg->rkcg_rk            = rk;
-        rkcg->rkcg_group_id      = group_id;
-        rkcg->rkcg_client_id     = client_id;
-        rkcg->rkcg_coord_id      = -1;
-        rkcg->rkcg_generation_id = -1;
-        rkcg->rkcg_wait_resp     = -1;
+        rkcg->rkcg_rk             = rk;
+        rkcg->rkcg_group_protocol = group_protocol;
+        rkcg->rkcg_group_id       = group_id;
+        rkcg->rkcg_client_id      = client_id;
+        rkcg->rkcg_coord_id       = -1;
+        rkcg->rkcg_generation_id  = -1;
+        rkcg->rkcg_wait_resp      = -1;
 
         rkcg->rkcg_ops                      = rd_kafka_q_new(rk);
         rkcg->rkcg_ops->rkq_serve           = rd_kafka_cgrp_op_serve;
@@ -420,6 +424,8 @@ rd_kafka_cgrp_t *rd_kafka_cgrp_new(rd_kafka_t *rk,
         rkcg->rkcg_q                        = rd_kafka_consume_q_new(rk);
         rkcg->rkcg_group_instance_id =
             rd_kafkap_str_new(rk->rk_conf.group_instance_id, -1);
+        rkcg->rkcg_group_remote_assignor =
+            rd_kafkap_str_new(rk->rk_conf.group_remote_assignor, -1);
 
         TAILQ_INIT(&rkcg->rkcg_topics);
         rd_list_init(&rkcg->rkcg_toppars, 32, NULL);
@@ -1513,8 +1519,8 @@ static void rd_kafka_cgrp_handle_SyncGroup_memberstate(
         const rd_kafka_topic_partition_field_t fields[] = {
             RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
             RD_KAFKA_TOPIC_PARTITION_FIELD_END};
-        if (!(assignment =
-                  rd_kafka_buf_read_topic_partitions(rkbuf, 0, fields)))
+        if (!(assignment = rd_kafka_buf_read_topic_partitions(
+                  rkbuf, rd_false /* don't use topic_id */, 0, fields)))
                 goto err_parse;
         rd_kafka_buf_read_kbytes(rkbuf, &UserData);
 
@@ -1814,8 +1820,8 @@ static int rd_kafka_group_MemberMetadata_consumer_read(
             RD_KAFKA_TOPIC_PARTITION_FIELD_PARTITION,
             RD_KAFKA_TOPIC_PARTITION_FIELD_END};
         if (Version >= 1 &&
-            !(rkgm->rkgm_owned =
-                  rd_kafka_buf_read_topic_partitions(rkbuf, 0, fields)))
+            !(rkgm->rkgm_owned = rd_kafka_buf_read_topic_partitions(
+                  rkbuf, rd_false /* don't use topic_id */, 0, fields)))
                 goto err;
 
         if (Version >= 2) {
@@ -5567,6 +5573,11 @@ rd_kafka_consumer_group_metadata(rd_kafka_t *rk) {
         return cgmetadata;
 }
 
+const char *rd_kafka_consumer_group_metadata_member_id(
+    const rd_kafka_consumer_group_metadata_t *group_metadata) {
+        return group_metadata->member_id;
+}
+
 void rd_kafka_consumer_group_metadata_destroy(
     rd_kafka_consumer_group_metadata_t *cgmetadata) {
         rd_free(cgmetadata->group_id);

src/rdkafka_cgrp.h

@@ -190,6 +190,12 @@ typedef struct rd_kafka_cgrp_s {
         int32_t rkcg_coord_id; /**< Current coordinator id,
                                 *   or -1 if not known. */
 
+        rd_kafka_group_protocol_t
+            rkcg_group_protocol; /**< Group protocol to use */
+
+        rd_kafkap_str_t *rkcg_group_remote_assignor; /**< Group remote
+                                                      *   assignor to use */
+
         rd_kafka_broker_t *rkcg_curr_coord; /**< Current coordinator
                                              *   broker handle, or NULL.
                                              *   rkcg_coord's nodename is
@@ -313,6 +319,7 @@ extern const char *rd_kafka_cgrp_join_state_names[];
 
 void rd_kafka_cgrp_destroy_final(rd_kafka_cgrp_t *rkcg);
 rd_kafka_cgrp_t *rd_kafka_cgrp_new(rd_kafka_t *rk,
+                                   rd_kafka_group_protocol_t group_protocol,
                                    const rd_kafkap_str_t *group_id,
                                    const rd_kafkap_str_t *client_id);
 void rd_kafka_cgrp_serve(rd_kafka_cgrp_t *rkcg);

src/rdkafka_conf.c

@@ -1132,9 +1132,26 @@ static const struct rd_kafka_property rd_kafka_properties[] = {
      "Group session keepalive heartbeat interval.", 1, 3600 * 1000, 3 * 1000},
     {_RK_GLOBAL | _RK_CGRP, "group.protocol.type", _RK_C_KSTR,
      _RK(group_protocol_type),
-     "Group protocol type. NOTE: Currently, the only supported group "
+     "Group protocol type for the `generic` group protocol. NOTE: Currently, "
+     "the only supported group "
      "protocol type is `consumer`.",
      .sdef = "consumer"},
+    {_RK_GLOBAL | _RK_CGRP | _RK_HIGH | _RK_HIDDEN, "group.protocol", _RK_C_S2I,
+     _RK(group_protocol),
+     "Group protocol to use. Use `generic` for the original protocol and "
+     "`consumer` for the new "
+     "protocol introduced in KIP-848. Available protocols: generic or "
+     "consumer. Default is `generic`, "
+     "but will change to `consumer` in next releases.",
+     .vdef = RD_KAFKA_GROUP_PROTOCOL_GENERIC,
+     .s2i  = {{RD_KAFKA_GROUP_PROTOCOL_GENERIC, "generic"},
+             {RD_KAFKA_GROUP_PROTOCOL_CONSUMER, "consumer"}}},
+    {_RK_GLOBAL | _RK_CGRP | _RK_MED | _RK_HIDDEN, "group.remote.assignor",
+     _RK_C_STR, _RK(group_remote_assignor),
+     "Server side assignor to use. Keep it null to make server select a "
+     "suitable assignor for the group. "
+     "Available assignors: uniform or range. Default is null",
+     .sdef = NULL},
     {_RK_GLOBAL | _RK_CGRP, "coordinator.query.interval.ms", _RK_C_INT,
      _RK(coord_query_intvl_ms),
      "How often to query for the current client group coordinator. "

src/rdkafka_conf.h

@@ -163,6 +163,11 @@ typedef enum {
         RD_KAFKA_RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY,
 } rd_kafka_client_dns_lookup_t;
 
+typedef enum {
+        RD_KAFKA_GROUP_PROTOCOL_GENERIC,
+        RD_KAFKA_GROUP_PROTOCOL_CONSUMER,
+} rd_kafka_group_protocol_t;
+
 /* Increase in steps of 64 as needed.
  * This must be larger than sizeof(rd_kafka_[topic_]conf_t) */
 #define RD_KAFKA_CONF_PROPS_IDX_MAX (64 * 33)
@@ -363,8 +368,10 @@ struct rd_kafka_conf_s {
         int fetch_min_bytes;
         int fetch_queue_backoff_ms;
         int fetch_error_backoff_ms;
+        rd_kafka_group_protocol_t group_protocol;
         char *group_id_str;
         char *group_instance_id;
+        char *group_remote_assignor;
         int allow_auto_create_topics;
 
         rd_kafka_pattern_list_t *topic_blacklist;