krabka_client_admin/

group_descriptions.rs

//! Batched group descriptions: [`AdminClient::describe_consumer_groups`],
//! [`AdminClient::describe_classic_groups`],
//! [`AdminClient::describe_share_groups`] and
//! [`AdminClient::describe_streams_groups`].
//!
//! Each call finds the coordinator of every group with one batched
//! `FindCoordinator` request and sends one describe request to each
//! coordinator, as Apache Kafka's `Describe*GroupsHandler` classes do with
//! their `CoordinatorStrategy`. A request to a coordinator that
//! fails or loses its connection finds the coordinator again.

use std::{
    collections::{BTreeMap, BTreeSet},
    sync::{Mutex, PoisonError},
};

use bytes::{Buf, BufMut};
use krabka_client_core::{ClientError, CoordinatorEndpoint};
use krabka_protocol::{
    Decode, Encode, ProtocolError, ProtocolRequest,
    owned::{
        common::consumer_group_describe_response::assignment::Assignment,
        consumer_group_describe_request::{self, ConsumerGroupDescribeRequest},
        consumer_group_describe_response::{
            DescribedGroup as ConsumerDescribedGroup, Member as ConsumerMember,
        },
        consumer_protocol_assignment::{self, ConsumerProtocolAssignment},
        describe_groups_request::DescribeGroupsRequest,
        describe_groups_response::{DescribedGroup as ClassicDescribedGroup, DescribedGroupMember},
        share_group_describe_request::ShareGroupDescribeRequest,
        share_group_describe_response::DescribedGroup as ShareDescribedGroup,
        streams_group_describe_request::{self, StreamsGroupDescribeRequest},
        streams_group_describe_response::{
            DescribedGroup as StreamsDescribedGroup, StreamsGroupDescribeResponse,
        },
    },
};

use crate::{
    AdminClient, AdminError, ClusterNode, KafkaError,
    cluster::{authorized_operations, node_with_host_port},
    group_coordinators::{GroupOutcome, group_error_outcome, kafka_error},
    groups::{GroupState, GroupType, list_groups_kafka_error},
    kafka_error_name,
    retry::is_connection_failure,
    users::AclOperation,
};

/// `UNKNOWN_SERVER_ERROR`: the code that Kafka's `ApiError.fromThrowable`
/// gives an exception with no Kafka error code.
const UNKNOWN_SERVER_ERROR: i16 = -1;
/// `UNSUPPORTED_VERSION`.
const UNSUPPORTED_VERSION: i16 = 35;
/// `GROUP_ID_NOT_FOUND`.
const GROUP_ID_NOT_FOUND: i16 = 69;
/// The protocol type of a classic consumer group
/// (`ConsumerProtocol.PROTOCOL_TYPE`).
const CONSUMER_PROTOCOL_TYPE: &str = "consumer";

/// A classic group's description from `DescribeGroups`, as Apache Kafka's
/// `ClassicGroupDescription` holds it. This is the broker's wire response
/// entry for the group.
pub type ClassicGroupDescription = ClassicDescribedGroup;

/// A share group's description from `ShareGroupDescribe`, as Apache Kafka's
/// `ShareGroupDescription` holds it. This is the broker's wire response entry
/// for the group.
pub type ShareGroupDescription = ShareDescribedGroup;

/// A streams group's description from `StreamsGroupDescribe`, as Apache
/// Kafka's `StreamsGroupDescription` holds it. This is the broker's wire
/// response entry for the group.
pub type StreamsGroupDescription = StreamsDescribedGroup;

/// The options of the describe calls, as Apache Kafka's
/// `Describe*GroupsOptions` hold them.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, bon::Builder)]
pub struct DescribeGroupsOptions {
    /// Ask for the operations that the caller may perform on each group.
    /// Kafka's default is `false`.
    #[builder(default)]
    pub include_authorized_operations: bool,
}

/// One member of a consumer group, as Apache Kafka's `MemberDescription`
/// holds it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemberDescription {
    pub member_id: String,
    pub group_instance_id: Option<String>,
    /// `None` for a member of a classic group.
    pub rack_id: Option<String>,
    pub client_id: String,
    pub host: String,
    /// The partitions that the member owns now.
    pub assignment: BTreeSet<(String, i32)>,
    /// The partitions that the coordinator wants the member to own. `None`
    /// for a member of a classic group.
    pub target_assignment: Option<BTreeSet<(String, i32)>>,
    /// `None` for a member of a classic group.
    pub member_epoch: Option<i32>,
    /// Whether the member uses the consumer protocol (KIP-1099). `None` when
    /// the coordinator does not say (`ConsumerGroupDescribe` v0) or for a
    /// member of a classic group.
    pub upgraded: Option<bool>,
}

/// A consumer group, classic or KIP-848, as Apache Kafka's
/// `ConsumerGroupDescription` holds it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConsumerGroupDescription {
    pub group_id: String,
    /// A classic group with no protocol type.
    pub is_simple_consumer_group: bool,
    pub members: Vec<MemberDescription>,
    /// The server assignor of a consumer group, or the protocol of a classic
    /// group.
    pub partition_assignor: String,
    /// [`GroupType::Consumer`] or [`GroupType::Classic`].
    pub group_type: GroupType,
    pub group_state: GroupState,
    /// The coordinator that `FindCoordinator` named.
    pub coordinator: ClusterNode,
    /// `None` when the caller did not ask for them.
    pub authorized_operations: Option<BTreeSet<AclOperation>>,
    /// `None` for a classic group.
    pub group_epoch: Option<i32>,
    /// `None` for a classic group.
    pub target_assignment_epoch: Option<i32>,
}

/// The per-group results of a describe call: one entry for each requested
/// group.
pub type GroupDescriptions<T> = BTreeMap<String, Result<T, KafkaError>>;

/// A `StreamsGroupDescribe` request limited to v0.
///
/// Apache Kafka 4.3.1 defines `StreamsGroupDescribe` v0 only. v1
/// (`IncludeTopologyDescription`, KIP-1331) is newer than that release, so
/// this type keeps version negotiation at v0.
#[derive(Clone, Debug, PartialEq, Eq)]
struct ReleasedStreamsGroupDescribe(StreamsGroupDescribeRequest);

impl Encode for ReleasedStreamsGroupDescribe {
    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
        self.0.encode(buf, version)
    }

    fn encoded_len(&self, version: i16) -> usize {
        self.0.encoded_len(version)
    }
}

impl ProtocolRequest for ReleasedStreamsGroupDescribe {
    const API_KEY: i16 = streams_group_describe_request::API_KEY;
    const MIN_VERSION: i16 = streams_group_describe_request::MIN_VERSION;
    /// The last `StreamsGroupDescribe` version of Kafka 4.3.1.
    const MAX_VERSION: i16 = 0;
    const LATEST_STABLE_VERSION: i16 = Self::MAX_VERSION;
    const FLEXIBLE_MIN: i16 = streams_group_describe_request::FLEXIBLE_MIN;
    type Response = StreamsGroupDescribeResponse;
}

trait GroupDescribeRequest: ProtocolRequest {
    fn for_groups(groups: Vec<String>, options: DescribeGroupsOptions) -> Self;
}

macro_rules! impl_group_describe_request {
    ($request:ty, $groups_field:ident) => {
        impl GroupDescribeRequest for $request {
            fn for_groups(groups: Vec<String>, options: DescribeGroupsOptions) -> Self {
                Self {
                    $groups_field: groups,
                    include_authorized_operations: options.include_authorized_operations,
                    ..Default::default()
                }
            }
        }
    };
}

impl_group_describe_request!(ConsumerGroupDescribeRequest, group_ids);
impl_group_describe_request!(DescribeGroupsRequest, groups);
impl_group_describe_request!(ShareGroupDescribeRequest, group_ids);

impl GroupDescribeRequest for ReleasedStreamsGroupDescribe {
    fn for_groups(groups: Vec<String>, options: DescribeGroupsOptions) -> Self {
        Self(StreamsGroupDescribeRequest {
            group_ids: groups,
            include_authorized_operations: options.include_authorized_operations,
            ..Default::default()
        })
    }
}

trait GroupDescriptionEntry: Sized {
    fn group_id(&self) -> &str;
    fn error_code(&self) -> i16;
    fn error_message(self) -> Option<String>;
    fn successful_outcome(self) -> GroupOutcome<Self>;

    fn into_outcome(self) -> (String, GroupOutcome<Self>) {
        let group_id = self.group_id().to_owned();
        let error_code = self.error_code();
        let outcome = if error_code == 0 {
            self.successful_outcome()
        } else {
            group_error_outcome(error_code, self.error_message(), &[])
        };
        (group_id, outcome)
    }
}

macro_rules! impl_group_description_entry {
    ($entry:ty, $error_message:expr, $successful_outcome:expr) => {
        impl GroupDescriptionEntry for $entry {
            fn group_id(&self) -> &str {
                &self.group_id
            }

            fn error_message(self) -> Option<String> {
                ($error_message)(self)
            }

            fn successful_outcome(self) -> GroupOutcome<Self> {
                ($successful_outcome)(self)
            }

            fn error_code(&self) -> i16 {
                self.error_code
            }
        }
    };
}

impl_group_description_entry!(
    ClassicDescribedGroup,
    |_: ClassicDescribedGroup| None,
    |group: ClassicDescribedGroup| GroupOutcome::Done(Ok(group))
);
impl_group_description_entry!(
    ShareDescribedGroup,
    |group: ShareDescribedGroup| group.error_message,
    |group: ShareDescribedGroup| GroupOutcome::Done(Ok(group))
);
impl_group_description_entry!(
    StreamsDescribedGroup,
    |group: StreamsDescribedGroup| group.error_message,
    |group: StreamsDescribedGroup| {
        if group.topology.is_none() {
            GroupOutcome::Done(Err(KafkaError {
                code: UNKNOWN_SERVER_ERROR,
                name: kafka_error_name(UNKNOWN_SERVER_ERROR),
                message: Some("Topology information is missing".to_owned()),
            }))
        } else {
            GroupOutcome::Done(Ok(group))
        }
    }
);

trait GroupDescribeApi: GroupDescribeRequest {
    type Description;

    const METHOD_NAME: &'static str;

    fn outcomes(response: Self::Response) -> BTreeMap<String, GroupOutcome<Self::Description>>;
}

macro_rules! impl_group_describe_api {
    ($request:ty, $description:ty, $method_name:literal) => {
        impl GroupDescribeApi for $request {
            type Description = $description;

            const METHOD_NAME: &'static str = $method_name;

            fn outcomes(
                response: Self::Response,
            ) -> BTreeMap<String, GroupOutcome<Self::Description>> {
                group_description_outcomes(response.groups)
            }
        }
    };
}

impl_group_describe_api!(
    DescribeGroupsRequest,
    ClassicGroupDescription,
    "describeClassicGroups"
);
impl_group_describe_api!(
    ShareGroupDescribeRequest,
    ShareGroupDescription,
    "describeShareGroups"
);
impl_group_describe_api!(
    ReleasedStreamsGroupDescribe,
    StreamsGroupDescription,
    "describeStreamsGroups"
);

fn group_description_outcomes<Group: GroupDescriptionEntry>(
    groups: impl IntoIterator<Item = Group>,
) -> BTreeMap<String, GroupOutcome<Group>> {
    groups
        .into_iter()
        .map(GroupDescriptionEntry::into_outcome)
        .collect()
}

fn insert_description_outcomes<Group, Description>(
    outcomes: &mut BTreeMap<String, GroupOutcome<Description>>,
    groups: impl IntoIterator<Item = Group>,
    mut classify: impl FnMut(Group) -> (String, Option<GroupOutcome<Description>>),
) {
    for group in groups {
        let (group_id, outcome) = classify(group);
        if let Some(outcome) = outcome {
            outcomes.insert(group_id, outcome);
        }
    }
}

fn consumer_description_outcome(
    group: ConsumerDescribedGroup,
    node: &ClusterNode,
    classic_keys: &mut Vec<String>,
    classic: &Mutex<BTreeMap<String, Option<String>>>,
) -> (String, Option<GroupOutcome<ConsumerGroupDescription>>) {
    let group_id = group.group_id.clone();
    let outcome = match consumer_answer(group, node) {
        ConsumerAnswer::Outcome(outcome) => Some(outcome),
        ConsumerAnswer::Classic { not_found_message } => {
            classic
                .lock()
                .unwrap_or_else(PoisonError::into_inner)
                .insert(group_id.clone(), not_found_message);
            classic_keys.push(group_id.clone());
            None
        }
    };
    (group_id, outcome)
}

fn classic_consumer_description_outcome(
    group: ClassicDescribedGroup,
    node: &ClusterNode,
    classic: &Mutex<BTreeMap<String, Option<String>>>,
) -> (String, Option<GroupOutcome<ConsumerGroupDescription>>) {
    let group_id = group.group_id.clone();
    let not_found_message = classic
        .lock()
        .unwrap_or_else(PoisonError::into_inner)
        .get(&group_id)
        .cloned()
        .flatten();
    (
        group_id,
        Some(classic_consumer_outcome(group, node, not_found_message)),
    )
}

/// The node that `FindCoordinator` named, as Kafka's `Node` holds it.
fn coordinator_node(coordinator: &CoordinatorEndpoint) -> ClusterNode {
    node_with_host_port(
        coordinator.node_id,
        coordinator.host.clone(),
        coordinator.port,
    )
}

/// The outcome of every group of `keys` after a request to their coordinator
/// failed with `error`: a failed or lost connection finds the coordinator
/// again, as Kafka's `AdminApiDriver.onFailure` does after a disconnect, and
/// every other failure is final.
pub(crate) fn request_error_outcomes<V>(
    keys: &[String],
    error: &AdminError,
) -> BTreeMap<String, GroupOutcome<V>> {
    keys.iter()
        .map(|key| {
            let outcome = if is_connection_failure(error) {
                GroupOutcome::Unmap(error.to_string())
            } else {
                GroupOutcome::Done(Err(list_groups_kafka_error(error)))
            };
            (key.clone(), outcome)
        })
        .collect()
}

/// The partitions of a `ConsumerGroupDescribe` assignment.
fn consumer_assignment(assignment: Assignment) -> BTreeSet<(String, i32)> {
    assignment_partitions(
        assignment
            .topic_partitions
            .into_iter()
            .map(|topic| (topic.topic_name, topic.partitions)),
    )
}

fn assignment_partitions(
    topics: impl IntoIterator<Item = (String, Vec<i32>)>,
) -> BTreeSet<(String, i32)> {
    topics
        .into_iter()
        .fold(BTreeSet::new(), |mut result, (name, partitions)| {
            result.extend(
                partitions
                    .into_iter()
                    .map(|partition| (name.clone(), partition)),
            );
            result
        })
}

/// A member of a `ConsumerGroupDescribe` answer, as Kafka's
/// `DescribeConsumerGroupsHandler.handledConsumerGroupResponse` reads it.
fn consumer_member(member: ConsumerMember) -> MemberDescription {
    MemberDescription {
        member_id: member.member_id,
        group_instance_id: member.instance_id,
        rack_id: member.rack_id,
        client_id: member.client_id,
        host: member.client_host,
        assignment: consumer_assignment(member.assignment),
        target_assignment: Some(consumer_assignment(member.target_assignment)),
        member_epoch: Some(member.member_epoch),
        upgraded: (member.member_type != -1).then_some(member.member_type == 1),
    }
}

/// Reads a version-prefixed `ConsumerProtocolAssignment`, as Kafka's
/// `ConsumerProtocol.deserializeAssignment` does: a version above the
/// highest known version reads as the highest known version, and a negative
/// version is an error.
fn classic_assignment(mut bytes: &[u8]) -> Result<BTreeSet<(String, i32)>, String> {
    if bytes.is_empty() {
        return Ok(BTreeSet::new());
    }
    if bytes.len() < 2 {
        return Err("Buffer underflow while parsing consumer protocol's header".to_owned());
    }
    let version = bytes.get_i16();
    if version < consumer_protocol_assignment::MIN_VERSION {
        return Err(format!("Unsupported assignment version: {version}"));
    }
    let version = version.min(consumer_protocol_assignment::MAX_VERSION);
    let assignment = ConsumerProtocolAssignment::decode(&mut bytes, version)
        .map_err(|error| format!("Error reading consumer assignment: {error}"))?;
    Ok(assignment_partitions(
        assignment
            .assigned_partitions
            .into_iter()
            .map(|topic| (topic.topic, topic.partitions)),
    ))
}

/// A member of a classic `DescribeGroups` answer, as Kafka's
/// `DescribeConsumerGroupsHandler.handledClassicGroupResponse` reads it.
fn classic_member(member: DescribedGroupMember) -> Result<MemberDescription, String> {
    Ok(MemberDescription {
        assignment: classic_assignment(&member.member_assignment)?,
        member_id: member.member_id,
        group_instance_id: member.group_instance_id,
        rack_id: None,
        client_id: member.client_id,
        host: member.client_host,
        target_assignment: None,
        member_epoch: None,
        upgraded: None,
    })
}

/// What the answer of `ConsumerGroupDescribe` means for one group.
#[derive(Debug, Clone, PartialEq, Eq)]
enum ConsumerAnswer {
    Outcome(GroupOutcome<ConsumerGroupDescription>),
    /// Describe the group with `DescribeGroups`, as Kafka's
    /// `useClassicGroupApi` set records it. The message of a
    /// `GROUP_ID_NOT_FOUND` answer, if any, is kept for the classic answer.
    Classic {
        not_found_message: Option<String>,
    },
}

/// Classifies one `ConsumerGroupDescribe` entry, as Kafka's
/// `DescribeConsumerGroupsHandler.handleError` does with
/// `isConsumerGroupResponse` set: `UNSUPPORTED_VERSION` and
/// `GROUP_ID_NOT_FOUND` move the group to `DescribeGroups`.
fn consumer_answer(group: ConsumerDescribedGroup, coordinator: &ClusterNode) -> ConsumerAnswer {
    match group.error_code {
        0 => ConsumerAnswer::Outcome(GroupOutcome::Done(Ok(ConsumerGroupDescription {
            group_id: group.group_id,
            is_simple_consumer_group: false,
            members: group.members.into_iter().map(consumer_member).collect(),
            partition_assignor: group.assignor_name,
            group_type: GroupType::Consumer,
            group_state: GroupState::parse(&group.group_state),
            coordinator: coordinator.clone(),
            authorized_operations: authorized_operations(group.authorized_operations),
            group_epoch: Some(group.group_epoch),
            target_assignment_epoch: Some(group.assignment_epoch),
        }))),
        UNSUPPORTED_VERSION => ConsumerAnswer::Classic {
            not_found_message: None,
        },
        GROUP_ID_NOT_FOUND => ConsumerAnswer::Classic {
            not_found_message: group.error_message,
        },
        code => ConsumerAnswer::Outcome(group_error_outcome(code, group.error_message, &[])),
    }
}

/// Classifies one `DescribeGroups` entry of `describe_consumer_groups`, as
/// Kafka's `DescribeConsumerGroupsHandler` does for a group in
/// `useClassicGroupApi`. A group whose protocol type is neither `consumer`
/// nor empty is not a consumer group. `GROUP_ID_NOT_FOUND` carries the
/// message of the `ConsumerGroupDescribe` answer when there was one.
fn classic_consumer_outcome(
    group: ClassicDescribedGroup,
    coordinator: &ClusterNode,
    not_found_message: Option<String>,
) -> GroupOutcome<ConsumerGroupDescription> {
    match group.error_code {
        0 => {}
        GROUP_ID_NOT_FOUND => {
            return GroupOutcome::Done(Err(kafka_error(
                GROUP_ID_NOT_FOUND,
                not_found_message.or(group.error_message),
            )));
        }
        code => return group_error_outcome(code, group.error_message, &[]),
    }
    if group.protocol_type != CONSUMER_PROTOCOL_TYPE && !group.protocol_type.is_empty() {
        return GroupOutcome::Done(Err(kafka_error(
            UNKNOWN_SERVER_ERROR,
            Some(format!(
                "GroupId {} is not a consumer group ({}).",
                group.group_id, group.protocol_type
            )),
        )));
    }
    let members = match group
        .members
        .into_iter()
        .map(classic_member)
        .collect::<Result<Vec<_>, _>>()
    {
        Ok(members) => members,
        Err(message) => {
            return GroupOutcome::Done(Err(kafka_error(UNKNOWN_SERVER_ERROR, Some(message))));
        }
    };
    GroupOutcome::Done(Ok(ConsumerGroupDescription {
        is_simple_consumer_group: group.protocol_type.is_empty(),
        group_id: group.group_id,
        members,
        partition_assignor: group.protocol_data,
        group_type: GroupType::Classic,
        group_state: GroupState::parse(&group.group_state),
        coordinator: coordinator.clone(),
        authorized_operations: authorized_operations(group.authorized_operations),
        group_epoch: None,
        target_assignment_epoch: None,
    }))
}

impl AdminClient {
    /// Describes consumer groups, classic and KIP-848, as Apache Kafka's
    /// `describeConsumerGroups` does with its `DescribeConsumerGroupsHandler`.
    ///
    /// The call finds the coordinator of every group with one batched
    /// `FindCoordinator` request, and sends one `ConsumerGroupDescribe`
    /// request to each coordinator with the groups that it coordinates. A
    /// group that the answer gives `GROUP_ID_NOT_FOUND` (69) or
    /// `UNSUPPORTED_VERSION` (35), and every group of a coordinator that does
    /// not support `ConsumerGroupDescribe`, is described with `DescribeGroups`
    /// instead, as Kafka's handler falls back for a classic group. Kafka
    /// sends that `DescribeGroups` request in the next round of its driver.
    /// This call sends it in the same round, to the same coordinator.
    ///
    /// `COORDINATOR_LOAD_IN_PROGRESS` (14) sends a group to the same
    /// coordinator again, and `COORDINATOR_NOT_AVAILABLE` (15) and
    /// `NOT_COORDINATOR` (16) find its coordinator again. A lost
    /// connection finds the coordinator again. The call waits with Kafka's
    /// backoff between rounds and stops at `default.api.timeout.ms` (60 s),
    /// where each pending group gets `REQUEST_TIMED_OUT` (7).
    ///
    /// # Errors
    /// The call itself does not fail. Each group gets its own [`KafkaError`]:
    /// the error code of the answer, `GROUP_ID_NOT_FOUND` with the message of
    /// the `ConsumerGroupDescribe` answer for a group that does not exist,
    /// `UNKNOWN_SERVER_ERROR` (-1) for a classic group that is not a consumer
    /// group, or `REQUEST_TIMED_OUT` (7) at the deadline.
    pub async fn describe_consumer_groups(
        &self,
        groups: &[&str],
        options: DescribeGroupsOptions,
    ) -> GroupDescriptions<ConsumerGroupDescription> {
        let classic = Mutex::new(BTreeMap::<String, Option<String>>::new());
        self.call_group_coordinators("describeConsumerGroups", groups, |coordinator, keys| {
            self.describe_consumer_groups_on(coordinator, keys, options, &classic)
        })
        .await
    }

    /// One `describe_consumer_groups` request to one coordinator. `classic`
    /// holds the groups that use `DescribeGroups`, with the message of their
    /// `GROUP_ID_NOT_FOUND` answer.
    async fn describe_consumer_groups_on(
        &self,
        coordinator: CoordinatorEndpoint,
        keys: Vec<String>,
        options: DescribeGroupsOptions,
        classic: &Mutex<BTreeMap<String, Option<String>>>,
    ) -> Result<BTreeMap<String, GroupOutcome<ConsumerGroupDescription>>, AdminError> {
        let connection = self.connect_coordinator(&coordinator).await?;
        let node = coordinator_node(&coordinator);
        let (classic_keys, consumer_keys): (Vec<_>, Vec<_>) = {
            let classic = classic.lock().unwrap_or_else(PoisonError::into_inner);
            keys.into_iter().partition(|key| classic.contains_key(key))
        };
        let mut outcomes = BTreeMap::new();
        let mut classic_keys = classic_keys;
        // Kafka's `NodeApiVersions` fails a request for an API that the
        // coordinator does not advertise with an `UnsupportedVersionException`,
        // and `handleUnsupportedVersionException` moves each group to the
        // classic API.
        let consumer_describe_supported = connection
            .advertised_api_range(consumer_group_describe_request::API_KEY)
            .is_some();
        if !consumer_describe_supported {
            let mut classic = classic.lock().unwrap_or_else(PoisonError::into_inner);
            for key in &consumer_keys {
                classic.entry(key.clone()).or_insert(None);
            }
            drop(classic);
            classic_keys.extend(consumer_keys.iter().cloned());
        } else if !consumer_keys.is_empty() {
            let request = ConsumerGroupDescribeRequest::for_groups(consumer_keys.clone(), options);
            match connection.send(request).await {
                Ok(response) => {
                    insert_description_outcomes(&mut outcomes, response.groups, |group| {
                        consumer_description_outcome(group, &node, &mut classic_keys, classic)
                    });
                }
                // Kafka's `handleUnsupportedVersionException` moves each group
                // to the classic API.
                Err(ClientError::IncompatibleVersion { .. }) => {
                    let mut classic = classic.lock().unwrap_or_else(PoisonError::into_inner);
                    for key in &consumer_keys {
                        classic.entry(key.clone()).or_insert(None);
                    }
                    drop(classic);
                    classic_keys.extend(consumer_keys);
                }
                Err(error) => {
                    let error = AdminError::from(error);
                    if is_connection_failure(&error) {
                        return Err(error);
                    }
                    outcomes.extend(request_error_outcomes(&consumer_keys, &error));
                }
            }
        }
        if classic_keys.is_empty() {
            return Ok(outcomes);
        }
        let request = DescribeGroupsRequest::for_groups(classic_keys.clone(), options);
        match connection.send(request).await {
            Ok(response) => {
                insert_description_outcomes(&mut outcomes, response.groups, |group| {
                    classic_consumer_description_outcome(group, &node, classic)
                });
            }
            Err(error) => {
                outcomes.extend(request_error_outcomes(&classic_keys, &error.into()));
            }
        }
        Ok(outcomes)
    }

    /// Describes classic groups with `DescribeGroups`, as Apache Kafka's
    /// `describeClassicGroups` does with its `DescribeClassicGroupsHandler`.
    ///
    /// The call finds the coordinator of every group with one batched
    /// `FindCoordinator` request and sends one `DescribeGroups` request to
    /// each coordinator. The coordinator codes retry as
    /// [`AdminClient::describe_consumer_groups`] describes. Every other code
    /// is the final error of its group, with no message, as the handler uses
    /// `error.message()`.
    ///
    /// # Errors
    /// The call itself does not fail. Each group gets its own [`KafkaError`].
    pub async fn describe_classic_groups(
        &self,
        groups: &[&str],
        options: DescribeGroupsOptions,
    ) -> GroupDescriptions<ClassicGroupDescription> {
        self.describe_group_api::<DescribeGroupsRequest>(groups, options)
            .await
    }

    /// Describes share groups with `ShareGroupDescribe`, as Apache Kafka's
    /// `describeShareGroups` does with its `DescribeShareGroupsHandler`.
    ///
    /// The call finds the coordinator of every group with one batched
    /// `FindCoordinator` request and sends one `ShareGroupDescribe` request
    /// to each coordinator. The coordinator codes retry as
    /// [`AdminClient::describe_consumer_groups`] describes. Every other code,
    /// such as `GROUP_ID_NOT_FOUND`, is the final error of its group, with
    /// the message of the answer.
    ///
    /// # Errors
    /// The call itself does not fail. Each group gets its own [`KafkaError`].
    pub async fn describe_share_groups(
        &self,
        groups: &[&str],
        options: DescribeGroupsOptions,
    ) -> GroupDescriptions<ShareGroupDescription> {
        self.describe_group_api::<ShareGroupDescribeRequest>(groups, options)
            .await
    }

    /// Describes streams groups with `StreamsGroupDescribe` v0, as Apache
    /// Kafka's `describeStreamsGroups` does with its
    /// `DescribeStreamsGroupsHandler`.
    ///
    /// The call finds the coordinator of every group with one batched
    /// `FindCoordinator` request and sends one `StreamsGroupDescribe` request
    /// to each coordinator. The coordinator codes retry as
    /// [`AdminClient::describe_consumer_groups`] describes. Every other code
    /// is the final error of its group, with the message of the answer. A
    /// group whose answer has no topology gets `UNKNOWN_SERVER_ERROR` (-1),
    /// as the handler fails it with an `IllegalStateException`.
    ///
    /// # Errors
    /// The call itself does not fail. Each group gets its own [`KafkaError`].
    pub async fn describe_streams_groups(
        &self,
        groups: &[&str],
        options: DescribeGroupsOptions,
    ) -> GroupDescriptions<StreamsGroupDescription> {
        self.describe_group_api::<ReleasedStreamsGroupDescribe>(groups, options)
            .await
    }

    async fn describe_group_api<Request: GroupDescribeApi>(
        &self,
        groups: &[&str],
        options: DescribeGroupsOptions,
    ) -> GroupDescriptions<Request::Description> {
        self.call_group_coordinators(
            Request::METHOD_NAME,
            groups,
            |coordinator, keys| async move {
                let connection = self.connect_coordinator(&coordinator).await?;
                let request = Request::for_groups(keys.clone(), options);
                Ok(match connection.send(request).await {
                    Ok(response) => Request::outcomes(response),
                    Err(error) => request_error_outcomes(&keys, &error.into()),
                })
            },
        )
        .await
    }
}

/// Classifies one `StreamsGroupDescribe` entry, as Kafka's
/// `DescribeStreamsGroupsHandler.handleResponse` does.
#[cfg(test)]
mod tests {
    use std::{
        net::SocketAddr,
        sync::{Arc, Mutex},
    };

    use assert2::assert;
    use bytes::BytesMut;
    use krabka_client_core::MockReply;
    use krabka_protocol::owned::{
        common::consumer_group_describe_response::topic_partitions::TopicPartitions,
        consumer_group_describe_response::ConsumerGroupDescribeResponse,
        consumer_protocol_assignment::TopicPartition as AssignedTopic, describe_groups_request,
        describe_groups_response::DescribeGroupsResponse, share_group_describe_request,
        share_group_describe_response::ShareGroupDescribeResponse,
        streams_group_describe_response::Topology,
    };

    use super::*;
    use crate::{
        group_coordinators::test_support::{GroupCluster, Lookup, Seen, group_cluster},
        partition_leaders::test_support::{ApiVersionRange, decode_request, encode_response},
    };

    fn api_range(api_key: i16, max_version: i16) -> ApiVersionRange {
        ApiVersionRange::builder()
            .api_key(api_key)
            .max_version(max_version)
            .build()
    }

    fn mock_response(bytes: Vec<u8>) -> MockReply {
        MockReply::Respond(bytes)
    }

    fn classic_response(groups: Vec<ClassicDescribedGroup>) -> DescribeGroupsResponse {
        DescribeGroupsResponse {
            groups,
            ..DescribeGroupsResponse::default()
        }
    }

    const FALLBACK_GROUPS: [&str; 5] = ["kip848", "classic", "other", "missing", "connect"];
    const STREAMS_GROUPS: [&str; 2] = ["with", "without"];

    struct AdminGroupCluster {
        cluster: GroupCluster,
        admin: AdminClient,
    }

    macro_rules! described_group {
        ($entry:ident, $error_code:expr, $error_message:expr $(, $field:ident = $value:expr)* $(,)?) => {
            $entry {
                error_code: $error_code,
                error_message: $error_message,
                group_id: "workers".to_owned(),
                $($field: $value,)*
                ..Default::default()
            }
        };
    }

    macro_rules! encode_described_group {
        ($response:ident, $entry:ident, $code:expr, $message:expr, $version:expr, $flexible:expr $(, $field:ident = $value:expr)* $(,)?) => {
            crate::encode_test_struct_field!(
                encode_response,
                $response,
                groups,
                vec![described_group!($entry, $code, $message $(, $field = $value)*)],
                $version,
                $flexible,
            )
        };
    }

    #[bon::builder]
    async fn admin_group_cluster(
        apis: Vec<ApiVersionRange>,
        answer: impl FnMut(usize, i16, i16, &[u8], SocketAddr) -> MockReply + Send + 'static,
    ) -> AdminGroupCluster {
        let cluster = group_cluster()
            .count(1)
            .apis(apis)
            .lookup(|_, _| Lookup::At(0))
            .answer(answer)
            .call()
            .await;
        let admin = cluster.admin(krabka_units::secs(5)).await;
        AdminGroupCluster { cluster, admin }
    }

    const OMITTED: i32 = i32::MIN;

    fn assignment_bytes(version: i16, partitions: &[(&str, &[i32])]) -> bytes::Bytes {
        let mut bytes = BytesMut::new();
        bytes.put_i16(version);
        ConsumerProtocolAssignment {
            assigned_partitions: partitions
                .iter()
                .map(|(topic, partitions)| AssignedTopic {
                    topic: (*topic).to_owned(),
                    partitions: partitions.to_vec(),
                    ..Default::default()
                })
                .collect(),
            ..Default::default()
        }
        .encode(&mut bytes, version.min(3))
        .expect("assignment encodes");
        bytes.freeze()
    }

    #[yare::parameterized(
        empty = { bytes::Bytes::new(), Ok(BTreeSet::new()) },
        v0 = { assignment_bytes(0, &[("orders", &[0, 2])]), Ok(maplit::btreeset! {
            ("orders".to_owned(), 0),
            ("orders".to_owned(), 2),
        }) },
        newer_version = { assignment_bytes(9, &[("orders", &[1])]), Ok(maplit::btreeset! {
            ("orders".to_owned(), 1),
        }) },
        negative_version = { bytes::Bytes::from_static(&[0xff, 0xff]), Err("Unsupported assignment version: -1".to_owned()) },
        truncated_header = { bytes::Bytes::from_static(&[0]), Err("Buffer underflow while parsing consumer protocol's header".to_owned()) },
    )]
    fn classic_assignment_reads_as_kafka_does(
        bytes: bytes::Bytes,
        expected: Result<BTreeSet<(String, i32)>, String>,
    ) {
        assert2::assert!(classic_assignment(&bytes) == expected);
    }

    fn node(addr: SocketAddr, id: i32) -> ClusterNode {
        node_with_host_port(id, addr.ip().to_string(), i32::from(addr.port()))
    }

    fn nodes(addrs: [SocketAddr; 2]) -> [ClusterNode; 2] {
        [node(addrs[0], 0), node(addrs[1], 1)]
    }

    fn consumer_entry(
        group_id: &str,
        error_code: i16,
        message: Option<&str>,
    ) -> ConsumerDescribedGroup {
        ConsumerDescribedGroup {
            error_code,
            error_message: message.map(str::to_owned),
            group_id: group_id.to_owned(),
            group_state: "Stable".to_owned(),
            group_epoch: 4,
            assignment_epoch: 3,
            assignor_name: "uniform".to_owned(),
            members: vec![ConsumerMember {
                member_id: "m1".to_owned(),
                client_id: "c1".to_owned(),
                client_host: "/10.0.0.1".to_owned(),
                member_epoch: 4,
                assignment: Assignment {
                    topic_partitions: vec![TopicPartitions {
                        topic_name: "orders".to_owned(),
                        partitions: vec![0],
                        ..Default::default()
                    }],
                    ..Default::default()
                },
                member_type: 1,
                ..Default::default()
            }],
            authorized_operations: OMITTED,
            ..Default::default()
        }
    }

    fn classic_entry(
        group_id: &str,
        error_code: i16,
        protocol_type: &str,
    ) -> ClassicDescribedGroup {
        ClassicDescribedGroup {
            error_code,
            group_id: group_id.to_owned(),
            group_state: "Stable".to_owned(),
            protocol_type: protocol_type.to_owned(),
            protocol_data: "range".to_owned(),
            members: vec![DescribedGroupMember {
                member_id: "m2".to_owned(),
                group_instance_id: Some("static-1".to_owned()),
                client_id: "c2".to_owned(),
                client_host: "/10.0.0.2".to_owned(),
                member_assignment: assignment_bytes(1, &[("orders", &[1])]),
                ..Default::default()
            }],
            authorized_operations: 1 << 3,
            ..Default::default()
        }
    }

    struct GroupDescriptionFields {
        member: MemberDescription,
        partition_assignor: String,
        group_type: GroupType,
        authorized_operations: Option<BTreeSet<AclOperation>>,
        group_epoch: Option<i32>,
        target_assignment_epoch: Option<i32>,
    }

    #[bon::builder]
    fn group_description(
        group_id: String,
        coordinator: ClusterNode,
        fields: GroupDescriptionFields,
    ) -> ConsumerGroupDescription {
        ConsumerGroupDescription {
            group_id,
            is_simple_consumer_group: false,
            members: vec![fields.member],
            partition_assignor: fields.partition_assignor,
            group_type: fields.group_type,
            group_state: GroupState::Stable,
            coordinator,
            authorized_operations: fields.authorized_operations,
            group_epoch: fields.group_epoch,
            target_assignment_epoch: fields.target_assignment_epoch,
        }
    }

    #[derive(Clone, Copy)]
    enum GroupDescriptionKind {
        Consumer,
        Classic,
    }

    #[bon::builder]
    fn member_description(
        member_id: String,
        group_instance_id: Option<String>,
        client_id: String,
        host: String,
        assignment: BTreeSet<(String, i32)>,
        target_assignment: Option<BTreeSet<(String, i32)>>,
        member_epoch: Option<i32>,
        upgraded: Option<bool>,
    ) -> MemberDescription {
        MemberDescription {
            member_id,
            group_instance_id,
            rack_id: None,
            client_id,
            host,
            assignment,
            target_assignment,
            member_epoch,
            upgraded,
        }
    }

    fn group_description_fields(kind: GroupDescriptionKind) -> GroupDescriptionFields {
        match kind {
            GroupDescriptionKind::Consumer => GroupDescriptionFields {
                member: member_description()
                    .member_id("m1".to_owned())
                    .maybe_group_instance_id(None)
                    .client_id("c1".to_owned())
                    .host("/10.0.0.1".to_owned())
                    .assignment(maplit::btreeset! {
                        ("orders".to_owned(), 0),
                    })
                    .maybe_target_assignment(Some(BTreeSet::new()))
                    .maybe_member_epoch(Some(4))
                    .maybe_upgraded(Some(true))
                    .call(),
                partition_assignor: "uniform".to_owned(),
                group_type: GroupType::Consumer,
                authorized_operations: None,
                group_epoch: Some(4),
                target_assignment_epoch: Some(3),
            },
            GroupDescriptionKind::Classic => GroupDescriptionFields {
                member: member_description()
                    .member_id("m2".to_owned())
                    .maybe_group_instance_id(Some("static-1".to_owned()))
                    .client_id("c2".to_owned())
                    .host("/10.0.0.2".to_owned())
                    .assignment(maplit::btreeset! {
                        ("orders".to_owned(), 1),
                    })
                    .maybe_target_assignment(None)
                    .maybe_member_epoch(None)
                    .maybe_upgraded(None)
                    .call(),
                partition_assignor: "range".to_owned(),
                group_type: GroupType::Classic,
                authorized_operations: Some(maplit::btreeset! {
                    AclOperation::Read,
                }),
                group_epoch: None,
                target_assignment_epoch: None,
            },
        }
    }

    fn test_group_description(
        group_id: &str,
        coordinator: ClusterNode,
        kind: GroupDescriptionKind,
    ) -> ConsumerGroupDescription {
        let fields = group_description_fields(kind);
        group_description()
            .group_id(group_id.to_owned())
            .coordinator(coordinator)
            .fields(fields)
            .call()
    }

    fn classic_group_description(coordinator: ClusterNode) -> ConsumerGroupDescription {
        test_group_description("classic", coordinator, GroupDescriptionKind::Classic)
    }

    #[derive(Clone, Copy, Debug)]
    enum DescribeApiCall {
        Classic,
        Share,
        Streams,
    }

    async fn describe_one_group(
        call: DescribeApiCall,
        admin: &AdminClient,
        options: DescribeGroupsOptions,
    ) -> Option<Result<(), KafkaError>> {
        match call {
            DescribeApiCall::Classic => admin
                .describe_classic_groups(&["workers"], options)
                .await
                .remove("workers")
                .map(|result| result.map(drop)),
            DescribeApiCall::Share => admin
                .describe_share_groups(&["workers"], options)
                .await
                .remove("workers")
                .map(|result| result.map(drop)),
            DescribeApiCall::Streams => admin
                .describe_streams_groups(&["workers"], options)
                .await
                .remove("workers")
                .map(|result| result.map(drop)),
        }
    }

    fn default_describe_options() -> DescribeGroupsOptions {
        DescribeGroupsOptions::builder().build()
    }

    async fn default_streams_descriptions(
        admin: &AdminClient,
        groups: &[&str],
    ) -> GroupDescriptions<StreamsGroupDescription> {
        admin
            .describe_streams_groups(groups, default_describe_options())
            .await
    }

    /// The group ids of each request of `api_key` that `seen` recorded, with
    /// the broker index and the version, in broker order.
    fn requests_of<T>(
        seen: &Seen,
        api_key: i16,
        read: impl Fn(&[u8], i16) -> T,
    ) -> Vec<(usize, i16, T)> {
        let mut requests = seen
            .requests
            .iter()
            .filter(|request| request.api_key == api_key)
            .map(|request| {
                (
                    request.broker,
                    request.version,
                    read(&request.body, request.version),
                )
            })
            .collect::<Vec<_>>();
        // The coordinators answer at the same time, so order by broker.
        requests.sort_by_key(|(broker, _, _)| *broker);
        requests
    }

    fn consumer_describe_ids(body: &[u8], version: i16) -> Vec<String> {
        decode_request::<ConsumerGroupDescribeRequest>(body, version, true).group_ids
    }

    fn describe_groups_ids(body: &[u8], version: i16) -> Vec<String> {
        decode_request::<DescribeGroupsRequest>(body, version, version >= 5).groups
    }

    /// Kafka's `DescribeConsumerGroupsHandler` sends one batched
    /// `ConsumerGroupDescribe` to each coordinator, falls back to
    /// `DescribeGroups` for a group that the coordinator gives
    /// `GROUP_ID_NOT_FOUND` or `UNSUPPORTED_VERSION`, keeps the message of
    /// the first `GROUP_ID_NOT_FOUND`, and fails a classic group that is not a
    /// consumer group.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn describe_consumer_groups_batches_and_falls_back_to_describe_groups() {
        // "kip848" and "classic" live on broker 0, "other" and "missing" and
        // "connect" on broker 1.
        let cluster: GroupCluster = group_cluster()
            .count(2)
            .apis(vec![
                api_range(consumer_group_describe_request::API_KEY, 1),
                api_range(describe_groups_request::API_KEY, 5),
            ])
            .lookup(|group, _| Lookup::At(usize::from(!matches!(group, "kip848" | "classic"))))
            .answer(|_, api_key, version, body, _| {
                let reply = match api_key {
                    consumer_group_describe_request::API_KEY => {
                        let ids = consumer_describe_ids(body, version);
                        let groups = ids
                            .iter()
                            .map(|id| match id.as_str() {
                                "kip848" | "other" => consumer_entry(id, 0, None),
                                "classic" | "connect" => {
                                    consumer_entry(id, 69, Some("Group is not a consumer group."))
                                }
                                "missing" => {
                                    consumer_entry(id, 69, Some("Group missing does not exist."))
                                }
                                _ => consumer_entry(id, 35, None),
                            })
                            .collect();
                        crate::encode_test_struct_field!(
                            encode_response,
                            ConsumerGroupDescribeResponse,
                            groups,
                            groups,
                            version,
                            true
                        )
                    }
                    describe_groups_request::API_KEY => {
                        let ids = describe_groups_ids(body, version);
                        let groups = ids
                            .iter()
                            .map(|id| match id.as_str() {
                                "classic" => classic_entry(id, 0, "consumer"),
                                "connect" => classic_entry(id, 0, "connect"),
                                _ => classic_entry(id, 69, ""),
                            })
                            .collect();
                        encode_response(&classic_response(groups), version, version >= 5)
                    }
                    _ => return MockReply::Silent,
                };
                mock_response(reply)
            })
            .call()
            .await;
        let admin = cluster.admin(krabka_units::secs(5)).await;

        let result = admin
            .describe_consumer_groups(&FALLBACK_GROUPS, default_describe_options())
            .await;

        let [b0, b1] = nodes([cluster.addrs[0], cluster.addrs[1]]);
        let seen = cluster.stop();
        assert2::assert!(
            result
                == maplit::btreemap! {
                    "kip848".to_owned() => Ok(test_group_description(
                    "kip848",
                    b0.clone(),
                    GroupDescriptionKind::Consumer,
                )),
                    "classic".to_owned() => Ok(classic_group_description(b0)),
                    "other".to_owned() => Ok(test_group_description(
                    "other",
                    b1,
                    GroupDescriptionKind::Consumer,
                )),
                    "missing".to_owned() => Err(kafka_error(
                    69,
                    Some("Group missing does not exist.".to_owned())
                )),
                    "connect".to_owned() => Err(kafka_error(
                    -1,
                    Some("GroupId connect is not a consumer group (connect).".to_owned())
                )),
                }
        );
        let ids = |ids: &[&str]| ids.iter().map(|id| (*id).to_owned()).collect::<Vec<_>>();
        assert!(
            (
                seen.find_coordinator.clone(),
                requests_of(
                    &seen,
                    consumer_group_describe_request::API_KEY,
                    consumer_describe_ids
                ),
                requests_of(&seen, describe_groups_request::API_KEY, describe_groups_ids),
            ) == (
                vec![(
                    6,
                    ids(&["classic", "connect", "kip848", "missing", "other"])
                )],
                vec![
                    (0, 1, ids(&["classic", "kip848"])),
                    (1, 1, ids(&["connect", "missing", "other"])),
                ],
                vec![
                    (0, 5, ids(&["classic"])),
                    (1, 5, ids(&["connect", "missing"]))
                ],
            )
        );
    }

    /// A coordinator that does not support `ConsumerGroupDescribe` gets
    /// `DescribeGroups` for every group, as Kafka's
    /// `handleUnsupportedVersionException` moves them to the classic API.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn describe_consumer_groups_uses_describe_groups_on_an_old_coordinator() {
        let AdminGroupCluster { cluster, admin } = admin_group_cluster()
            .apis(vec![api_range(describe_groups_request::API_KEY, 5)])
            .answer(|_, api_key, version, body, _| {
                if api_key != describe_groups_request::API_KEY {
                    return MockReply::Silent;
                }
                let groups = describe_groups_ids(body, version)
                    .iter()
                    .map(|id| classic_entry(id, 0, "consumer"))
                    .collect();
                mock_response(encode_response(&classic_response(groups), version, true))
            })
            .call()
            .await;

        let result = admin
            .describe_consumer_groups(&["classic"], default_describe_options())
            .await;

        let first_broker = cluster.addrs[0];
        let _seen = cluster.stop();
        let b0 = node(first_broker, 0);
        assert!(
            result
                == maplit::btreemap! {
                    "classic".to_owned() => Ok(classic_group_description(b0)),
                }
        );
    }

    /// Kafka's `handleUnsupportedVersionException`: a coordinator that lists
    /// `ConsumerGroupDescribe` only at versions this client does not speak
    /// gets each group through `DescribeGroups`.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn describe_consumer_groups_falls_back_when_no_version_is_common() {
        let AdminGroupCluster { cluster, admin } = admin_group_cluster()
            .apis(vec![
                ApiVersionRange::builder()
                    .api_key(consumer_group_describe_request::API_KEY)
                    .min_version(50)
                    .max_version(50)
                    .build(),
                api_range(describe_groups_request::API_KEY, 5),
            ])
            .answer(|_, api_key, version, body, _| {
                // The fallback sends `DescribeGroups` only.
                assert!(api_key == describe_groups_request::API_KEY);
                let groups = describe_groups_ids(body, version)
                    .iter()
                    .map(|id| classic_entry(id, 0, "consumer"))
                    .collect();
                mock_response(encode_response(&classic_response(groups), version, true))
            })
            .call()
            .await;

        let result = admin
            .describe_consumer_groups(&["classic"], default_describe_options())
            .await;

        let first_broker = cluster.addrs[0];
        let _seen = cluster.stop();
        let b0 = node(first_broker, 0);
        assert!(
            result
                == maplit::btreemap! {
                    "classic".to_owned() => Ok(classic_group_description(b0)),
                }
        );
    }

    /// Every describe handler retries `COORDINATOR_LOAD_IN_PROGRESS` on the
    /// same coordinator, finds the coordinator again on
    /// `COORDINATOR_NOT_AVAILABLE` and `NOT_COORDINATOR`, and fails every
    /// other code with the message of the answer (none for
    /// `describeClassicGroups`). The call sends the options' flag.
    #[yare::parameterized(
        classic_no_error = { DescribeApiCall::Classic, "no error", vec![0], 1, 1, Ok(()) },
        classic_load = { DescribeApiCall::Classic, "load in progress", vec![14, 0], 1, 2, Ok(()) },
        classic_not_coordinator = { DescribeApiCall::Classic, "not coordinator", vec![16, 0], 2, 2, Ok(()) },
        classic_not_available = { DescribeApiCall::Classic, "not available", vec![15, 0], 2, 2, Ok(()) },
        classic_authorization = { DescribeApiCall::Classic, "group authorization failed", vec![30], 1, 1, Err(30) },
        classic_not_found = { DescribeApiCall::Classic, "group id not found", vec![69], 1, 1, Err(69) },
        share_no_error = { DescribeApiCall::Share, "no error", vec![0], 1, 1, Ok(()) },
        share_load = { DescribeApiCall::Share, "load in progress", vec![14, 0], 1, 2, Ok(()) },
        share_not_coordinator = { DescribeApiCall::Share, "not coordinator", vec![16, 0], 2, 2, Ok(()) },
        share_not_available = { DescribeApiCall::Share, "not available", vec![15, 0], 2, 2, Ok(()) },
        share_authorization = { DescribeApiCall::Share, "group authorization failed", vec![30], 1, 1, Err(30) },
        share_not_found = { DescribeApiCall::Share, "group id not found", vec![69], 1, 1, Err(69) },
        streams_no_error = { DescribeApiCall::Streams, "no error", vec![0], 1, 1, Ok(()) },
        streams_load = { DescribeApiCall::Streams, "load in progress", vec![14, 0], 1, 2, Ok(()) },
        streams_not_coordinator = { DescribeApiCall::Streams, "not coordinator", vec![16, 0], 2, 2, Ok(()) },
        streams_not_available = { DescribeApiCall::Streams, "not available", vec![15, 0], 2, 2, Ok(()) },
        streams_authorization = { DescribeApiCall::Streams, "group authorization failed", vec![30], 1, 1, Err(30) },
        streams_not_found = { DescribeApiCall::Streams, "group id not found", vec![69], 1, 1, Err(69) },
    )]
    #[test_macro(tokio::test(flavor = "multi_thread", worker_threads = 2))]
    async fn describe_calls_retry_coordinator_errors_as_kafka_does(
        call: DescribeApiCall,
        name: &'static str,
        codes: Vec<i16>,
        expected_lookups: usize,
        expected_requests: usize,
        expected: Result<(), i16>,
    ) {
        let message = Some("from the broker".to_owned());
        let api_key = match call {
            DescribeApiCall::Classic => describe_groups_request::API_KEY,
            DescribeApiCall::Share => share_group_describe_request::API_KEY,
            DescribeApiCall::Streams => streams_group_describe_request::API_KEY,
        };
        let script = Arc::new(Mutex::new(codes));
        let reply_message = message.clone();
        let cluster = group_cluster()
            .count(1)
            .apis(vec![api_range(api_key, 6)])
            .lookup(|_, _| Lookup::At(0))
            .answer(move |_, key, version, _, _| {
                if key != api_key {
                    return MockReply::Silent;
                }
                let code = {
                    let mut script = script.lock().expect("script lock");
                    if script.len() > 1 {
                        script.remove(0)
                    } else {
                        script[0]
                    }
                };
                let body = match call {
                    DescribeApiCall::Classic => encode_described_group!(
                        DescribeGroupsResponse,
                        ClassicDescribedGroup,
                        code,
                        reply_message.clone(),
                        version,
                        version >= 5,
                    ),
                    DescribeApiCall::Share => encode_described_group!(
                        ShareGroupDescribeResponse,
                        ShareDescribedGroup,
                        code,
                        reply_message.clone(),
                        version,
                        true,
                    ),
                    DescribeApiCall::Streams => encode_described_group!(
                        StreamsGroupDescribeResponse,
                        StreamsDescribedGroup,
                        code,
                        reply_message.clone(),
                        version,
                        true,
                        topology = Some(Topology::default()),
                    ),
                };
                mock_response(body)
            })
            .call()
            .await;
        let admin = cluster.admin(krabka_units::secs(5)).await;
        let options = DescribeGroupsOptions::builder()
            .include_authorized_operations(true)
            .build();
        let result = describe_one_group(call, &admin, options).await;
        let seen = cluster.stop();
        let expected_message = match call {
            DescribeApiCall::Classic => None,
            DescribeApiCall::Share | DescribeApiCall::Streams => message.clone(),
        };
        let flags = seen
            .requests
            .iter()
            .map(|request| {
                let (version, body) = (request.version, &request.body[..]);
                match call {
                    DescribeApiCall::Classic => {
                        decode_request::<DescribeGroupsRequest>(body, version, version >= 5)
                            .include_authorized_operations
                    }
                    DescribeApiCall::Share => {
                        decode_request::<ShareGroupDescribeRequest>(body, version, true)
                            .include_authorized_operations
                    }
                    DescribeApiCall::Streams => {
                        decode_request::<StreamsGroupDescribeRequest>(body, version, true)
                            .include_authorized_operations
                    }
                }
            })
            .collect::<Vec<_>>();
        assert2::assert!(
            (result, seen.find_coordinator.len(), flags)
                == (
                    Some(expected.map_err(|code| kafka_error(code, expected_message))),
                    expected_lookups,
                    vec![true; expected_requests],
                ),
            "case {call:?} {name}"
        );
    }

    /// Kafka 4.3.1 defines `StreamsGroupDescribe` v0 only, and its handler
    /// fails a group whose answer has no topology.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn describe_streams_groups_sends_v0_and_needs_a_topology() {
        let AdminGroupCluster { cluster, admin } = admin_group_cluster()
            .apis(vec![api_range(streams_group_describe_request::API_KEY, 1)])
            .answer(|_, api_key, version, _, _| {
                if api_key != streams_group_describe_request::API_KEY {
                    return MockReply::Silent;
                }
                mock_response(crate::encode_test_struct_field!(
                    encode_response,
                    StreamsGroupDescribeResponse,
                    groups,
                    vec![
                        StreamsDescribedGroup {
                            group_id: "with".to_owned(),
                            topology: Some(Topology::default()),
                            ..Default::default()
                        },
                        StreamsDescribedGroup {
                            group_id: "without".to_owned(),
                            ..Default::default()
                        },
                    ],
                    version,
                    true,
                ))
            })
            .call()
            .await;

        let result = default_streams_descriptions(&admin, &STREAMS_GROUPS).await;

        let seen = cluster.stop();
        let versions = seen
            .requests
            .iter()
            .map(|request| request.version)
            .collect::<Vec<_>>();
        assert!(
            (result, versions)
                == (
                    maplit::btreemap! {
                        "with".to_owned() => Ok(StreamsDescribedGroup {
                        group_id: "with".to_owned(),
                        topology: Some(Topology::default()),
                        ..Default::default()
                    }),
                        "without".to_owned() => Err(kafka_error(
                        -1,
                        Some("Topology information is missing".to_owned())
                    )),
                    },
                    vec![0],
                )
        );
    }
}