admin: add DescribeLogDirs admin API (KIP-113) (#5192)

Implement the DescribeLogDirs admin API (Kafka API key 35) to allow
querying partition sizes in bytes on disk for capacity planning and
monitoring.

Adds full C admin API following the ElectLeaders pattern with
self-contained protocol request/response. Supports protocol versions
0-4 with flexible version negotiation (v2+). The topics parameter
accepts a topic-partition list to filter, or NULL for all topics.
Request targets a specific broker via AdminOptions_set_broker(),
defaults to controller. No fanout: single broker per call, same as
Java AdminClient.

Author: piochelepiotr

CHANGELOG.md

@@ -3,6 +3,7 @@
 librdkafka v2.14.0 is a feature release:
 
 * [KIP-768](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=186877575#KIP768:ExtendSASL/OAUTHBEARERwithSupportforOIDC-ClientConfiguration) Extend SASL/OAUTHBEARER to support OIDC claim mapping beyond the default `sub` claim (#5336).
+* [KIP-113](https://cwiki.apache.org/confluence/display/KAFKA/KIP-113%3A+Support+replicas+movement+between+log+directories) Add support for AdminAPI `DescribeLogDirs()` (#5333, @piochelepiotr).
 
 # librdkafka v2.13.2
 

src/rdkafka.h

@@ -5610,6 +5610,8 @@ typedef int rd_kafka_event_type_t;
 #define RD_KAFKA_EVENT_LISTOFFSETS_RESULT 0x400000
 /** ElectLeaders_result_t */
 #define RD_KAFKA_EVENT_ELECTLEADERS_RESULT 0x800000
+/** DescribeLogDirs_result_t */
+#define RD_KAFKA_EVENT_DESCRIBELOGDIRS_RESULT 0x1000000
 
 /**
  * @returns the event type for the given event.
@@ -5894,6 +5896,8 @@ typedef rd_kafka_event_t rd_kafka_AlterUserScramCredentials_result_t;
 typedef rd_kafka_event_t rd_kafka_ListOffsets_result_t;
 /*! ElectLeaders result type */
 typedef rd_kafka_event_t rd_kafka_ElectLeaders_result_t;
+/*! DescribeLogDirs result type */
+typedef rd_kafka_event_t rd_kafka_DescribeLogDirs_result_t;
 
 /**
  * @brief Get CreateTopics result.
@@ -7118,6 +7122,7 @@ typedef enum rd_kafka_admin_op_t {
         RD_KAFKA_ADMIN_OP_DESCRIBECLUSTER, /**< DescribeCluster */
         RD_KAFKA_ADMIN_OP_LISTOFFSETS,     /**< ListOffsets */
         RD_KAFKA_ADMIN_OP_ELECTLEADERS,    /**< ElectLeaders */
+        RD_KAFKA_ADMIN_OP_DESCRIBELOGDIRS, /**< DescribeLogDirs */
         RD_KAFKA_ADMIN_OP__CNT             /**< Number of ops defined */
 } rd_kafka_admin_op_t;
 
@@ -10140,6 +10145,190 @@ rd_kafka_ElectLeaders_result_partitions(
 
 /**@}*/
 
+/**
+ * @name Admin API - DescribeLogDirs
+ * @{
+ */
+
+/**
+ * @brief Opaque type for a log directory description.
+ */
+typedef struct rd_kafka_LogDirDescription_s rd_kafka_LogDirDescription_t;
+
+/**
+ * @brief Opaque type for a log directory topic description.
+ */
+typedef struct rd_kafka_LogDirTopicDescription_s
+    rd_kafka_LogDirTopicDescription_t;
+
+/**
+ * @brief Opaque type for a log directory partition description.
+ */
+typedef struct rd_kafka_LogDirPartitionDescription_s
+    rd_kafka_LogDirPartitionDescription_t;
+
+/**
+ * @brief Describe log directories on a broker.
+ *
+ * @param rk Client instance.
+ * @param topics Topic-partition list to filter results, or NULL
+ *        to describe all topics on the target broker.
+ * @param options Optional admin options, or NULL for defaults.
+ *                Valid options:
+ *                 - rd_kafka_AdminOptions_set_broker()
+ * @param rkqu Queue to emit result on.
+ *
+ * @remark The result event type emitted on the supplied queue is of type
+ *         \c RD_KAFKA_EVENT_DESCRIBELOGDIRS_RESULT
+ */
+RD_EXPORT void
+rd_kafka_DescribeLogDirs(rd_kafka_t *rk,
+                         const rd_kafka_topic_partition_list_t *topics,
+                         const rd_kafka_AdminOptions_t *options,
+                         rd_kafka_queue_t *rkqu);
+
+/**
+ * @brief Get an array of log dir descriptions from a
+ *        DescribeLogDirs result.
+ *
+ * @param result Result to get log dir descriptions from.
+ * @param cntp is updated to the number of elements in the array.
+ *
+ * @remark The lifetime of the returned memory is the same
+ *         as the lifetime of the \p result object.
+ */
+RD_EXPORT const rd_kafka_LogDirDescription_t **
+rd_kafka_DescribeLogDirs_result_descriptions(
+    const rd_kafka_DescribeLogDirs_result_t *result,
+    size_t *cntp);
+
+/**
+ * @brief Get DescribeLogDirs result.
+ *
+ * @returns the result of a DescribeLogDirs request, or NULL if
+ * event is of different type.
+ *
+ * @remark The lifetime of the returned memory is the same
+ *         as the lifetime of the \p rkev object.
+ *
+ * Event types:
+ *   RD_KAFKA_EVENT_DESCRIBELOGDIRS_RESULT
+ */
+RD_EXPORT const rd_kafka_DescribeLogDirs_result_t *
+rd_kafka_event_DescribeLogDirs_result(rd_kafka_event_t *rkev);
+
+/**
+ * @brief Get the error for a log dir description.
+ *
+ * @param logdir The log dir description.
+ *
+ * @returns the error associated with the log dir, or NULL
+ *          on success.
+ *
+ * @remark The lifetime of the returned memory is the same
+ *         as the lifetime of the \p logdir object.
+ */
+RD_EXPORT const rd_kafka_error_t *
+rd_kafka_LogDirDescription_error(const rd_kafka_LogDirDescription_t *logdir);
+
+/**
+ * @brief Get the log directory path from a log dir description.
+ *
+ * @param logdir The log dir description.
+ *
+ * @remark The lifetime of the returned memory is the same
+ *         as the lifetime of the \p logdir object.
+ */
+RD_EXPORT const char *
+rd_kafka_LogDirDescription_log_dir(const rd_kafka_LogDirDescription_t *logdir);
+
+/**
+ * @brief Get the array of topic descriptions from a log dir
+ *        description.
+ *
+ * @param logdir The log dir description.
+ * @param cntp is updated to the number of topics.
+ *
+ * @remark The lifetime of the returned memory is the same
+ *         as the lifetime of the \p logdir object.
+ */
+RD_EXPORT const rd_kafka_LogDirTopicDescription_t **
+rd_kafka_LogDirDescription_topics(const rd_kafka_LogDirDescription_t *logdir,
+                                  size_t *cntp);
+
+/**
+ * @brief Get the topic name from a log dir topic description.
+ *
+ * @param topic_desc The topic description.
+ *
+ * @remark The lifetime of the returned memory is the same
+ *         as the lifetime of the \p topic_desc object.
+ */
+RD_EXPORT const char *rd_kafka_LogDirTopicDescription_topic(
+    const rd_kafka_LogDirTopicDescription_t *topic_desc);
+
+/**
+ * @brief Get the array of partition descriptions from a log dir
+ *        topic description.
+ *
+ * @param topic_desc The topic description.
+ * @param cntp is updated to the number of partitions.
+ *
+ * @remark The lifetime of the returned memory is the same
+ *         as the lifetime of the \p topic_desc object.
+ */
+RD_EXPORT const rd_kafka_LogDirPartitionDescription_t **
+rd_kafka_LogDirTopicDescription_partitions(
+    const rd_kafka_LogDirTopicDescription_t *topic_desc,
+    size_t *cntp);
+
+/**
+ * @brief Get the partition index from a log dir partition
+ *        description.
+ *
+ * @param partition_desc The partition description.
+ *
+ * @returns the partition index.
+ */
+RD_EXPORT int32_t rd_kafka_LogDirPartitionDescription_partition(
+    const rd_kafka_LogDirPartitionDescription_t *partition_desc);
+
+/**
+ * @brief Get the partition size in bytes from a log dir partition
+ *        description.
+ *
+ * @param partition_desc The partition description.
+ *
+ * @returns the size in bytes on disk.
+ */
+RD_EXPORT int64_t rd_kafka_LogDirPartitionDescription_size(
+    const rd_kafka_LogDirPartitionDescription_t *partition_desc);
+
+/**
+ * @brief Get the offset lag from a log dir partition description.
+ *
+ * @param partition_desc The partition description.
+ *
+ * @returns the offset lag.
+ */
+RD_EXPORT int64_t rd_kafka_LogDirPartitionDescription_offset_lag(
+    const rd_kafka_LogDirPartitionDescription_t *partition_desc);
+
+/**
+ * @brief Get the is_future flag from a log dir partition
+ *        description.
+ *
+ * @param partition_desc The partition description.
+ *
+ * @returns 1 if this is a future log directory (i.e., the
+ *          partition is being moved to this log dir),
+ *          0 otherwise.
+ */
+RD_EXPORT int rd_kafka_LogDirPartitionDescription_is_future(
+    const rd_kafka_LogDirPartitionDescription_t *partition_desc);
+
+/**@}*/
+
 /**
  * @name Security APIs
  * @{