Additional AdminClient operations.

This PR adds three admin operations that were not previously supported:
- AdminClient::list_offsets: List offsets for the specified
topic_partitions.
- AdminClient::describe_consumer_groups: Provides a description on the
  requested consumer groups.
- AdminClient::list_consumer_group_offsets: List offset information for
  the topics subscribed by a consumer group.

Implementation notes:
- This mostly my first Rust code, so please comment and critique. The
work started because I was in search of something to do after having
read rust tutorials and documents. I decided for re-doing a python tool
I've recently coded for Kafka administration tasks and I found that some
of the operations I was using in python were missing from the admin
interface in rdkafka. Although the implementation I did in my rust tool
was different, I decided to try my hand at enhancing rdkafka and this PR
 is the result.
- For all three operations I've used a similar pattern for the result:
`KafkaResult<Vec<Result<TYPE, (KEY, KafkaError)>`. This is similar to
the existing GroupResult. The alternative was to insert the error value
in the TYPE record, following closer the librdkafka structure, but I
felt it was less in line with rust conventions.
- I felt that I was adding many types and that it was best to define
them in a different module/file for each operation. What I've done is
re-exporting them from the admin module.
- I've defined types::RDKafkaXXX for all the bindings::rd_kafka_xxxx
types, using the RDKafkaXXX alias in the code, keeping the function
calls to rdsys::rd_kafka_xxxx
- Regarding GroupResult and ConsumerGroupResult: Technically they both
come from the same rd_kafka_group_result_t, GroupResul just extracting
the consumer group name. The GroupResult type was already used by the
delete_groups operation. In order not to break the  existing API, I'm
maintaining both. Perhaps in the future they could be enhanced (or
perhaps it is well as it is, as the additional information is not
meaningful in the delete_groups case).
- Added use cases covering the different operations. I've probably not
been exhaustive, but I hope the coverage is enough.
- Documentation and field and structures descriptions I've taken from a
mixture of the java, python and the protocol documentation. Hope it
is in line with the rest of the documentation provided.

Author: j-santander

rdkafka-sys/src/types.rs

@@ -96,6 +96,39 @@ pub type RDKafkaTopicResult = bindings::rd_kafka_topic_result_t;
 /// Native rdkafka group result.
 pub type RDKafkaGroupResult = bindings::rd_kafka_group_result_t;
 
+/// Native rdkafka list offsets result.
+pub type RDKafkaListOffsetsResult = bindings::rd_kafka_ListOffsets_result_t;
+
+/// Native rdkafka list offsets info returned.
+pub type RDKafkaListOffsetsResultInfo = bindings::rd_kafka_ListOffsetsResultInfo_t;
+
+/// Native result of the Describe Consumer Group operation.
+pub type RDKafkaDescribeConsumerGroupsResult = bindings::rd_kafka_DescribeConsumerGroups_result_t;
+
+/// Native description of a Consumer Group.
+pub type RDKafkaConsumerGroupDescription = bindings::rd_kafka_ConsumerGroupDescription_t;
+
+/// Native ACL Operation enum.
+pub type RDKafkaAclOperation = bindings::rd_kafka_AclOperation_t;
+
+/// Native Consumer Group State enum.
+pub type RDKafkaConsumerGroupState = bindings::rd_kafka_consumer_group_state_t;
+
+/// Native Node data.
+pub type RDKafkaNode = bindings::rd_kafka_Node_t;
+
+/// Native Consumer Group Type enum.
+pub type RDKafkaConsumerGroupType = bindings::rd_kafka_consumer_group_type_t;
+
+/// Native Member Description data.
+pub type RDKafkaMemberDescription = bindings::rd_kafka_MemberDescription_t;
+
+/// Native Member Assignment data.
+pub type RDKafkaMemberAssignment = bindings::rd_kafka_MemberAssignment_t;
+
+/// Native information provided by the List Consumer Group Offsets operation.
+pub type RDKafkaListConsumerGroupOffsets = bindings::rd_kafka_ListConsumerGroupOffsets_t;
+
 /// Native rdkafka mock cluster.
 pub type RDKafkaMockCluster = bindings::rd_kafka_mock_cluster_t;
 

src/admin.rs

@@ -28,14 +28,26 @@ use crate::log::{trace, warn};
 use crate::util::{cstr_to_owned, AsCArray, ErrBuf, IntoOpaque, KafkaDrop, NativePtr, Timeout};
 use crate::TopicPartitionList;
 
+// Reexport the symbols defined in the auxiliary files
+pub use crate::group_description::{
+    group_description_result_key, AclOperation, ConsumerGroupDescription,
+    ConsumerGroupDescriptionResult, ConsumerGroupState, MemberAssignment, MemberDescription,
+};
+pub use crate::list_consumer_group_offsets::{
+    group_result_key, ConsumerGroup, ConsumerGroupResult, ListConsumerGroupOffsets,
+};
+pub use crate::list_offsets_result_info::{
+    list_offsets_result_key, ListOffsetsResult, ListOffsetsResultInfo,
+};
+
 //
 // ********** ADMIN CLIENT **********
 //
 
 /// A client for the Kafka admin API.
 ///
 /// `AdminClient` provides programmatic access to managing a Kafka cluster,
-/// notably manipulating topics, partitions, and configuration paramaters.
+/// notably manipulating topics, partitions, and configuration parameters.
 pub struct AdminClient<C: ClientContext> {
     client: Client<C>,
     queue: Arc<NativeQueue>,
@@ -377,6 +389,134 @@ impl<C: ClientContext> AdminClient<C> {
         Ok(rx)
     }
 
+    ///
+    /// List offsets for the specified topic_partitions. This operation enables to find the
+    /// beginning offset, end offset as well as the offset matching a timestamp in partitions or
+    /// the offset with max timestamp.
+    ///
+    /// In order to get the latest or the earliest offset, define the input [TopicPartitionListElem]
+    /// with [Offset::Beginning] or [Offset::End].
+    ///
+    pub fn list_offsets(
+        &self,
+        offsets: &TopicPartitionList,
+        opts: &AdminOptions,
+    ) -> impl Future<Output = KafkaResult<Vec<ListOffsetsResult>>> {
+        match self.list_offsets_inner(offsets, opts) {
+            Ok(rx) => Either::Left(ListOffsetsFuture { rx }),
+            Err(err) => Either::Right(future::err(err)),
+        }
+    }
+
+    fn list_offsets_inner(
+        &self,
+        offsets: &TopicPartitionList,
+        opts: &AdminOptions,
+    ) -> KafkaResult<oneshot::Receiver<NativeEvent>> {
+        let mut err_buf = ErrBuf::new();
+        let (native_opts, rx) = opts.to_native(self.client.native_ptr(), &mut err_buf)?;
+        unsafe {
+            rdsys::rd_kafka_ListOffsets(
+                self.client.native_ptr(),
+                offsets.ptr(),
+                native_opts.ptr(),
+                self.queue.ptr(),
+            );
+        }
+        Ok(rx)
+    }
+
+    ///
+    /// Provides a description on the requested consumer groups.
+    ///
+    pub fn describe_consumer_groups<'a, I>(
+        &self,
+        groups: I,
+        opts: &AdminOptions,
+    ) -> impl Future<Output = KafkaResult<Vec<ConsumerGroupDescriptionResult>>>
+    where
+        I: IntoIterator<Item = &'a &'a str>,
+    {
+        match self.describe_consumer_groups_inner(groups, opts) {
+            Ok(rx) => Either::Left(ConsumerGroupDescriptionFuture { rx }),
+            Err(err) => Either::Right(future::err(err)),
+        }
+    }
+
+    fn describe_consumer_groups_inner<'a, I>(
+        &self,
+        groups: I,
+        opts: &AdminOptions,
+    ) -> KafkaResult<oneshot::Receiver<NativeEvent>>
+    where
+        I: IntoIterator<Item = &'a &'a str>,
+    {
+        let mut native_groups = Vec::new();
+        for g in groups {
+            let cstring = CString::new(*g).expect("Failed to create CString");
+            native_groups.push(cstring.into_raw());
+        }
+
+        let mut err_buf = ErrBuf::new();
+        let (native_opts, rx) = opts.to_native(self.client.native_ptr(), &mut err_buf)?;
+        unsafe {
+            rdsys::rd_kafka_DescribeConsumerGroups(
+                self.client.native_ptr(),
+                native_groups.as_ptr() as *mut _,
+                native_groups.len(),
+                native_opts.ptr(),
+                self.queue.ptr(),
+            );
+        }
+        Ok(rx)
+    }
+
+    ///
+    /// List offset information for the consumer group and (optional) topic partition provided in
+    /// the request.
+    ///
+    /// Note that, while the API takes a vector as input, it will only support one group at a time.
+    pub fn list_consumer_group_offsets<'a, I>(
+        &self,
+        groups: I,
+        opts: &AdminOptions,
+    ) -> impl Future<Output = KafkaResult<Vec<ConsumerGroupResult>>>
+    where
+        I: IntoIterator<Item = &'a ListConsumerGroupOffsets<'a>>,
+    {
+        match self.list_consumer_group_offsets_inner(groups, opts) {
+            Ok(rx) => Either::Left(ListConsumerGroupOffsetsFuture { rx }),
+            Err(err) => Either::Right(future::err(err)),
+        }
+    }
+
+    fn list_consumer_group_offsets_inner<'a, I>(
+        &self,
+        groups: I,
+        opts: &AdminOptions,
+    ) -> KafkaResult<oneshot::Receiver<NativeEvent>>
+    where
+        I: IntoIterator<Item = &'a ListConsumerGroupOffsets<'a>>,
+    {
+        let mut native_groups = Vec::new();
+        for g in groups {
+            native_groups.push(g.to_native()?);
+        }
+
+        let mut err_buf = ErrBuf::new();
+        let (native_opts, rx) = opts.to_native(self.client.native_ptr(), &mut err_buf)?;
+        unsafe {
+            rdsys::rd_kafka_ListConsumerGroupOffsets(
+                self.client.native_ptr(),
+                native_groups.as_c_array(),
+                native_groups.len(),
+                native_opts.ptr(),
+                self.queue.ptr(),
+            );
+        }
+        Ok(rx)
+    }
+
     /// Returns the client underlying this admin client.
     pub fn inner(&self) -> &Client<C> {
         &self.client
@@ -1341,3 +1481,86 @@ impl Future for AlterConfigsFuture {
         Poll::Ready(Ok(out))
     }
 }
+
+//
+// List offsets handling
+//
+
+struct ListOffsetsFuture {
+    rx: oneshot::Receiver<NativeEvent>,
+}
+
+impl Future for ListOffsetsFuture {
+    type Output = KafkaResult<Vec<ListOffsetsResult>>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let event = ready!(self.rx.poll_unpin(cx)).map_err(|_| KafkaError::Canceled)?;
+        event.check_error()?;
+        let res = unsafe { rdsys::rd_kafka_event_ListOffsets_result(event.ptr()) };
+        if res.is_null() {
+            let typ = unsafe { rdsys::rd_kafka_event_type(event.ptr()) };
+            return Poll::Ready(Err(KafkaError::AdminOpCreation(format!(
+                "list offsets request received response of incorrect type ({})",
+                typ
+            ))));
+        }
+        let info_list =
+            unsafe { ListOffsetsResultInfo::vec_from_ptr(res as *mut RDKafkaListOffsetsResult) };
+        Poll::Ready(Ok(info_list))
+    }
+}
+
+//
+// Describe Consumer Groups handling
+//
+
+struct ConsumerGroupDescriptionFuture {
+    rx: oneshot::Receiver<NativeEvent>,
+}
+
+impl Future for ConsumerGroupDescriptionFuture {
+    type Output = KafkaResult<Vec<Result<ConsumerGroupDescription, (String, KafkaError)>>>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let event = ready!(self.rx.poll_unpin(cx)).map_err(|_| KafkaError::Canceled)?;
+        event.check_error()?;
+        let res = unsafe { rdsys::rd_kafka_event_DescribeConsumerGroups_result(event.ptr()) };
+        if res.is_null() {
+            let typ = unsafe { rdsys::rd_kafka_event_type(event.ptr()) };
+            return Poll::Ready(Err(KafkaError::AdminOpCreation(format!(
+                "describe consumer groups request received response of incorrect type ({})",
+                typ
+            ))));
+        }
+        let group_list = unsafe { ConsumerGroupDescription::vec_from_ptr(res) };
+        Poll::Ready(Ok(group_list))
+    }
+}
+
+//
+// List Consumer Group Offsets handling
+//
+struct ListConsumerGroupOffsetsFuture {
+    rx: oneshot::Receiver<NativeEvent>,
+}
+
+impl Future for ListConsumerGroupOffsetsFuture {
+    type Output = KafkaResult<Vec<ConsumerGroupResult>>;
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let event = ready!(self.rx.poll_unpin(cx)).map_err(|_| KafkaError::Canceled)?;
+        event.check_error()?;
+
+        let res = unsafe { rdsys::rd_kafka_event_ListConsumerGroupOffsets_result(event.ptr()) };
+        if res.is_null() {
+            let typ = unsafe { rdsys::rd_kafka_event_type(event.ptr()) };
+            return Poll::Ready(Err(KafkaError::AdminOpCreation(format!(
+                "list consumer group offsets request received response of incorrect type ({})",
+                typ
+            ))));
+        }
+
+        let mut n = 0;
+        let groups = unsafe { rdsys::rd_kafka_DeleteGroups_result_groups(res, &mut n) };
+        Poll::Ready(Ok(ConsumerGroup::vec_result_from_ptr(groups, n)))
+    }
+}