KIP 951 - Phase 1

src/rdkafka_buf.h

@@ -836,6 +836,29 @@ struct rd_kafka_buf_s { /* rd_kafka_buf_t */
                 }                                                              \
         } while (0)
 
+/**
+ * @brief Read KIP-482 Tags at current position in the buffer using
+ *        the `read_tag` function receiving the `opaque' pointer.
+ */
+#define rd_kafka_buf_read_tags(rkbuf, read_tag, opaque)                        \
+        do {                                                                   \
+                uint64_t _tagcnt;                                              \
+                if (!((rkbuf)->rkbuf_flags & RD_KAFKA_OP_F_FLEXVER))           \
+                        break;                                                 \
+                rd_kafka_buf_read_uvarint(rkbuf, &_tagcnt);                    \
+                while (_tagcnt-- > 0) {                                        \
+                        uint64_t _tagtype, _taglen;                            \
+                        rd_kafka_buf_read_uvarint(rkbuf, &_tagtype);           \
+                        rd_kafka_buf_read_uvarint(rkbuf, &_taglen);            \
+                        int _read_tag_resp =                                   \
+                            read_tag(rkbuf, _tagtype, _taglen, opaque);        \
+                        if (_read_tag_resp == -1)                              \
+                                goto err_parse;                                \
+                        if (!_read_tag_resp && _taglen > 0)                    \
+                                rd_kafka_buf_skip(rkbuf, (size_t)(_taglen));   \
+                }                                                              \
+        } while (0)
+
 /**
  * @brief Write empty tags at the current position in the buffer.
  */

src/rdkafka_proto.h

@@ -726,5 +726,31 @@ rd_kafka_pid_bump(const rd_kafka_pid_t old) {
 
 /**@}*/
 
+/**
+ * @name Current Leader and NodeEndpoints for KIP-951
+ *       response triggered metadata updates.
+ * @{
+ *
+ */
+
+typedef struct rd_kafkap_CurrentLeader_s {
+        int32_t LeaderId;
+        int32_t LeaderEpoch;
+} rd_kafkap_CurrentLeader_t;
+
+typedef struct rd_kafkap_NodeEndpoint_s {
+        int32_t NodeId;
+        rd_kafkap_str_t Host;
+        int32_t Port;
+        rd_kafkap_str_t Rack;
+} rd_kafkap_NodeEndpoint_t;
+
+typedef struct rd_kafkap_NodeEndpoints_s {
+        int32_t NodeEndpointCnt;
+        rd_kafkap_NodeEndpoint_t *NodeEndpoints;
+} rd_kafkap_NodeEndpoints_t;
+
+/**@}*/
+
 
 #endif /* _RDKAFKA_PROTO_H_ */

src/rdkafka_request.c

@@ -497,6 +497,59 @@ int rd_kafka_buf_write_topic_partitions(
 }
 
 
+/**
+ * @brief Read current leader from \p rkbuf.
+ *
+ * @param rkbuf buffer to read from
+ * @param CurrentLeader is the CurrentLeader to populate.
+ * @return 1 on success, else -1 on parse error.
+ */
+int rd_kafka_buf_read_CurrentLeader(rd_kafka_buf_t *rkbuf,
+                                    rd_kafkap_CurrentLeader_t *CurrentLeader) {
+        const int log_decode_errors = LOG_ERR;
+        rd_kafka_buf_read_i32(rkbuf, &CurrentLeader->LeaderId);
+        rd_kafka_buf_read_i32(rkbuf, &CurrentLeader->LeaderEpoch);
+        rd_kafka_buf_skip_tags(rkbuf);
+        return 1;
+err_parse:
+        return -1;
+}
+
+/**
+ * @brief Read NodeEndpoints from \p rkbuf.
+ *
+ * @param rkbuf buffer to read from
+ * @param NodeEndpoints is the NodeEndpoints to populate.
+ * @returns 1 on success, else -1 on parse error.
+ */
+int rd_kafka_buf_read_NodeEndpoints(rd_kafka_buf_t *rkbuf,
+                                    rd_kafkap_NodeEndpoints_t *NodeEndpoints) {
+        const int log_decode_errors = LOG_ERR;
+        int32_t i;
+        rd_kafka_buf_read_arraycnt(rkbuf, &NodeEndpoints->NodeEndpointCnt,
+                                   RD_KAFKAP_BROKERS_MAX);
+        RD_IF_FREE(NodeEndpoints->NodeEndpoints, rd_free);
+        NodeEndpoints->NodeEndpoints =
+            rd_calloc(NodeEndpoints->NodeEndpointCnt,
+                      sizeof(*NodeEndpoints->NodeEndpoints));
+
+        for (i = 0; i < NodeEndpoints->NodeEndpointCnt; i++) {
+                rd_kafka_buf_read_i32(rkbuf,
+                                      &NodeEndpoints->NodeEndpoints[i].NodeId);
+                rd_kafka_buf_read_str(rkbuf,
+                                      &NodeEndpoints->NodeEndpoints[i].Host);
+                rd_kafka_buf_read_i32(rkbuf,
+                                      &NodeEndpoints->NodeEndpoints[i].Port);
+                rd_kafka_buf_read_str(rkbuf,
+                                      &NodeEndpoints->NodeEndpoints[i].Rack);
+                rd_kafka_buf_skip_tags(rkbuf);
+        }
+        return 1;
+err_parse:
+        return -1;
+}
+
+
 /**
  * @brief Send FindCoordinatorRequest.
  *

src/rdkafka_request.h

@@ -99,6 +99,13 @@ int rd_kafka_buf_write_topic_partitions(
     rd_bool_t use_topic_name,
     const rd_kafka_topic_partition_field_t *fields);
 
+int rd_kafka_buf_read_CurrentLeader(rd_kafka_buf_t *rkbuf,
+                                    rd_kafkap_CurrentLeader_t *CurrentLeader);
+
+int rd_kafka_buf_read_NodeEndpoints(rd_kafka_buf_t *rkbuf,
+                                    rd_kafkap_NodeEndpoints_t *NodeEndpoints);
+
+
 rd_kafka_resp_err_t
 rd_kafka_FindCoordinatorRequest(rd_kafka_broker_t *rkb,
                                 rd_kafka_coordtype_t coordtype,