Method DescribeConsumerGroups

admin.go:1231–1269

(groups []string)

}

func (ca *clusterAdmin) DescribeConsumerGroups(groups []string) (result []*GroupDescription, err error) {
	groupsPerBroker := make(map[*Broker][]string)

	for _, group := range groups {
		coordinator, err := ca.client.Coordinator(group)
		if err != nil {
			return nil, err
		}
		groupsPerBroker[coordinator] = append(groupsPerBroker[coordinator], group)
	}

	for broker, brokerGroups := range groupsPerBroker {
		describeReq := &DescribeGroupsRequest{
			Groups: brokerGroups,
		}

		if ca.conf.Version.IsAtLeast(V2_4_0_0) {
			// Starting in version 4, the response will include group.instance.id info for members.
			// Starting in version 5, the response uses flexible encoding
			describeReq.Version = 5
		} else if ca.conf.Version.IsAtLeast(V2_3_0_0) {
			// Starting in version 3, authorized operations can be requested.
			describeReq.Version = 3
		} else if ca.conf.Version.IsAtLeast(V2_0_0_0) {
			// Version 2 is the same as version 0.
			describeReq.Version = 2
		} else if ca.conf.Version.IsAtLeast(V1_1_0_0) {
			// Version 1 is the same as version 0.
			describeReq.Version = 1
		}
		response, err := broker.DescribeGroups(describeReq)
		if err != nil {
			return nil, err
		}

		result = append(result, response.Groups...)
	}
	return result, nil
}

func (ca *clusterAdmin) ListConsumerGroups() (allGroups map[string]string, err error) {
	allGroups = make(map[string]string)