/**
    * Describe some classic groups in the cluster.
    *
    * @param groupIds The IDs of the groups to describe.
    * @param options  The options to use when describing the groups.
    * @return The DescribeClassicGroupsResult.
    */
   DescribeClassicGroupsResult describeClassicGroups(Collection<String> groupIds,
                                                     DescribeClassicGroupsOptions options);

   /**
    * Describe some classic groups in the cluster, with the default options.
    * <p>
    * This is a convenience method for {@link #describeClassicGroups(Collection, DescribeClassicGroupsOptions)}
    * with default options. See the overload for more details.
    *
    * @param groupIds The IDs of the groups to describe.
    * @return The DescribeClassicGroupsResult.
    */
   default DescribeClassicGroupsResult describeClassicGroups(Collection<String> groupIds) {
       return describeClassicGroups(groupIds, new DescribeClassicGroupsOptions());
   }

   /**
    * List the groups available in the cluster with the default options.
    *
    * <p>This is a convenience method for {@link #listGroups(ListGroupsOptions)} with default options.
    * See the overload for more details.
    *
    * @return The ListGroupsResult.
    */
   default ListGroupsResult listGroups() {
       return listGroups(new ListGroupsOptions());
   }
 
   /**
    * List the groups available in the cluster.
    *
    * @param options The options to use when listing the groups.
    * @return The ListGroupsResult.
    */
   ListGroupsResult listGroups(ListGroupsOptions options);

/**
 * Options for {@link Admin#describeClassicGroups(Collection, DescribeClassicGroupsOptions)}.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class DescribeClassicGroupsOptions extends AbstractOptions<DescribeClassicGroupsOptions> {
    private boolean includeAuthorizedOperations;

    public DescribeClassicGroupsOptions includeAuthorizedOperations(boolean includeAuthorizedOperations) {
        this.includeAuthorizedOperations = includeAuthorizedOperations;
        return this;
    }

    public boolean includeAuthorizedOperations() {
        return includeAuthorizedOperations;
    }
}

package org.apache.kafka.client.admin;

/**
 * The result of the {@link KafkaAdminClient#describeClassicGroups(Collection, DescribeClassicGroupsOptions)}} call.
 *
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class DescribeClassicGroupsResult {

    public DescribeClassicGroupsResult(final Map<String, KafkaFuture<ClassicGroupDescription>> futures);

    /**
     * Return a map from group id to futures which yield group descriptions.
     */
    public Map<String, KafkaFuture<ClassicGroupDescription>> describedGroups();

    /**
     * Return a future which yields all ClassicGroupDescription objects, if all the describes succeed.
     */
    public KafkaFuture<Map<String, ClassicGroupDescription>> all();
}

package org.apache.kafka.client.admin;

/**
 * A detailed description of a single classic group in the cluster.
 */
@InterfaceStability.Evolving
public class ClassicGroupDescription {
    public ClassicGroupDescription(String groupId,
                                   String protocol,
                                   Collection<MemberDescription> members,
                                   String partitionAssignor,
                                   GroupState groupState,
                                   Node coordinator);

    public ClassicGroupDescription(String groupId,
                                   String protocol,
                                   Collection<MemberDescription> members,
                                   String partitionAssignor,
                                   GroupState groupState,
                                   Node coordinator,
                                   Set<AclOperation> authorizedOperations);

    /**
     * The id of the classic group.
     */
    public String groupId();

    /**
     * The group protocol type.
     */
    public String protocol();

    /**
     * If the group is a simple consumer group or not.
     */
    public boolean isSimpleConsumerGroup();

    /**
     * A list of the members of the classic group.
     */
    public Collection<MemberDescription> members();

    /**
     * The group partition assignor.
     */
    public String partitionAssignor();

    /**
     * The group state, or UNKNOWN if the state is too new for us to parse.
     */
    public GroupState groupState();

    /**
     * The group coordinator, or null if the coordinator is not known.
     */
    public Node coordinator();

    /**
     * authorizedOperations for this group, or null if that information is not known.
     */
    public Set<AclOperation> authorizedOperations();
}

package org.apache.kafka.client.admin;
 
import org.apache.kafka.common.GroupType;
 
/**
 * Options for {@link Admin#listGroups(ListGroupsOptions)}.
 *
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class ListGroupsOptions extends AbstractOptions<ListGroupsOptions> {

    /**
     * In groupStates is set, only groups in these states will be returned by listGroups().
     * Otherwise, all groups are returned.
     * This operation is supported by brokers with version 2.6.0 or later.
     */
    public ListGroupsOptions inStates(Set<GroupState> groupStates) {
        this.groupStates = (groupStates == null || groupStates.isEmpty()) ? Collections.emptySet() : new HashSet<>(groupStates);
        return this;
    }

    /**
     * If types is set, only groups of these types will be returned by listGroups().
     * Otherwise, all groups are returned.
     */
    public ListGroupsOptions withTypes(Set<GroupType> types) {
        this.types = (types == null || types.isEmpty()) ? Collections.emptySet() : new HashSet<>(types);
        return this;
    }

    /**
     * Returns the list of group states that are requested or empty if not states have been specified.
     */
    public Set<GroupState> groupStates() {
        return groupStates;
    }

    /**
     * Returns the list of group types that are requested or empty if no types have been specified.
     */
    public Set<GroupType> types() {
        return types;
    }
}

package org.apache.kafka.clients.admin;
   
/**
 * The result of the {@link Admin#listGroups(ListGroupsOptions)} call.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class ListGroupsResult {
    ListGroupsResult(KafkaFuture<Collection<Object>> future) {
        super(future);
    }

    /**
     * Returns a future that yields either an exception, or the full set of group listings.
     */
    public KafkaFuture<Collection<GroupListing>> all() {
    }
  
    /**
     * Returns a future which yields just the valid listings.
     */
    public KafkaFuture<Collection<GroupListing>> valid() {
    }
   
    /**
     * Returns a future which yields just the errors which occurred.
     */
    public KafkaFuture<Collection<Throwable>> errors() {
    }
}

package org.apache.kafka.client.admin;
  
/**
 * A listing of a group in the cluster.
 * <p>
 * The API of this class is evolving, see {@link Admin} for details.
 */
@InterfaceStability.Evolving
public class GroupListing {
  public GroupListing(String groupId, Optional<GroupType> type, String protocol);

  /**
   * The id of the group.
   */
  public String groupId();
  
  /**
   * The group type.
   */
  public Optional<GroupType> type();
  
  /**
   * The group protocol type.
   */
  public String protocol();

  /**
   * The group state.
   */
  public Optional<GroupState> groupState();

  /**
   * If the group is a simple consumer group or not.
   */
  public boolean isSimpleConsumerGroup();
}

package org.apache.kafka.common;

/**
 * The group state.
 */
public enum GroupState {
    UNKNOWN("Unknown"),
    PREPARING_REBALANCE("PreparingRebalance"),
    COMPLETING_REBALANCE("CompletingRebalance"),
    STABLE("Stable"),
    DEAD("Dead"),
    EMPTY("Empty"),
    ASSIGNING("Assigning"),
    RECONCILING("Reconciling");

    GroupState(String name);

    /**
     * Case-insensitive group state lookup by string name.
     */
    public static GroupState parse(String name);

    public String toString();
}

{
  "apiKey": 15,
  "type": "request",
  "listeners": ["zkBroker", "broker"],
  "name": "DescribeGroupsRequest",
  // Versions 1 and 2 are the same as version 0.
  //
  // Starting in version 3, authorized operations can be requested.
  //
  // Starting in version 4, the response will include group.instance.id info for members.
  //
  // Version 5 is the first flexible version.
  //
  // Version 6 returns error code GROUP_ID_NOT_FOUND if the group ID is not found (KIP-1043).
  "validVersions": "0-6",
  "flexibleVersions": "6+",
  "fields": [
    { "name": "Groups", "type": "[]string", "versions": "0+", "entityType": "groupId",
      "about": "The names of the groups to describe" },
    { "name": "IncludeAuthorizedOperations", "type": "bool", "versions": "3+",
      "about": "Whether to include authorized operations." }
  ]
}

{
  "apiKey": 15,
  "type": "response",
  "name": "DescribeGroupsResponse",
  // Version 1 added throttle time.
  //
  // Starting in version 2, on quota violation, brokers send out responses before throttling.
  //
  // Starting in version 3, brokers can send authorized operations.
  //
  // Starting in version 4, the response will optionally include group.instance.id info for members.
  //
  // Version 5 is the first flexible version.
  //
  // Version 6 returns error code GROUP_ID_NOT_FOUND if the group ID is not found (KIP-1043).
  "validVersions": "0-6",
  "flexibleVersions": "6+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "1+", "ignorable": true,
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "Groups", "type": "[]DescribedGroup", "versions": "0+",
      "about": "Each described group.", "fields": [
      { "name": "ErrorCode", "type": "int16", "versions": "0+",
        "about": "The describe error, or 0 if there was no error." },
      { "name": "ErrorMessage", "type": "string", "versions": "6+", "nullableVersions": "6+", "default": "null",
        "about": "The describe error message, or null if there was no error." },
      { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
        "about": "The group ID string." },
      { "name": "GroupState", "type": "string", "versions": "0+",
        "about": "The group state string, or the empty string." },
      { "name": "ProtocolType", "type": "string", "versions": "0+",
        "about": "The group protocol type, or the empty string." },
      // ProtocolData is currently only filled in if the group state is in the Stable state.
      { "name": "ProtocolData", "type": "string", "versions": "0+",
        "about": "The group protocol data, or the empty string." },
      // N.B. If the group is in the Dead state, the members array will always be empty.
      { "name": "Members", "type": "[]DescribedGroupMember", "versions": "0+",
        "about": "The group members.", "fields": [
        { "name": "MemberId", "type": "string", "versions": "0+",
          "about": "The member ID assigned by the group coordinator." },
        { "name": "GroupInstanceId", "type": "string", "versions": "4+", "ignorable": true,
          "nullableVersions": "4+", "default": "null",
          "about": "The unique identifier of the consumer instance provided by end user." },
        { "name": "ClientId", "type": "string", "versions": "0+",
          "about": "The client ID used in the member's latest join group request." },
        { "name": "ClientHost", "type": "string", "versions": "0+",
          "about": "The client host." },
        // This is currently only provided if the group is in the Stable state.
        { "name": "MemberMetadata", "type": "bytes", "versions": "0+",
          "about": "The metadata corresponding to the current group protocol in use." },
        // This is currently only provided if the group is in the Stable state.
        { "name": "MemberAssignment", "type": "bytes", "versions": "0+",
          "about": "The current assignment provided by the group leader." }
      ]},
      { "name": "AuthorizedOperations", "type": "int32", "versions": "3+",  "default": "-2147483648",
        "about": "32-bit bitfield to represent authorized operations for this group." }
    ]}
  ]
}

$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --list
GROUP                   TYPE          PROTOCOL
old-consumer-group      Classic       consumer
new-consumer-group      Consumer      consumer
connect-cluster         Classic       connect
share-group             Share         share
schema-registry         Classic       sr
simple-consumer-group   Classic

$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --list --consumer
GROUP                   TYPE          PROTOCOL
old-consumer-group      Classic       consumer
new-consumer-group      Consumer      consumer
simple-consumer-group   Classic

$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --list --group-type consumer
GROUP                   TYPE          PROTOCOL
new-consumer-group      Consumer      consumer

$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --list --share
GROUP                   TYPE          PROTOCOL
share-group             Share         share

$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group SG1
Error: Group 'SG1' is not a consumer group.

$ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --describe --group CG1
Error: Group 'CG1' is not a share group.