/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.kafka.tools.consumer.group;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.ConsumerGroupDescription;
import org.apache.kafka.clients.admin.MemberDescription;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.GroupProtocol;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.RangeAssignor;
import org.apache.kafka.clients.consumer.RoundRobinAssignor;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.GroupState;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.GroupIdNotFoundException;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.test.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterConfigProperty;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTestDefaults;
import org.apache.kafka.common.test.api.Type;
import org.apache.kafka.common.utils.AppInfoParser;
import org.apache.kafka.common.utils.Exit;
import org.apache.kafka.test.TestUtils;
import org.apache.kafka.tools.ToolsTestUtils;

import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;
import java.util.stream.IntStream;
import java.util.stream.Stream;

import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.CONSUMER_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG;
import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.CONSUMER_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG;
import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG;
import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG;
import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG;
import static org.apache.kafka.test.TestUtils.RANDOM;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;

@ClusterTestDefaults(
    types = {Type.CO_KRAFT},
    serverProperties = {
        @ClusterConfigProperty(key = OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
        @ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),
        @ClusterConfigProperty(key = GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, value = "1000"),
        @ClusterConfigProperty(key = CONSUMER_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG, value = "500"),
        @ClusterConfigProperty(key = CONSUMER_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG, value = "500"),
    }
)
public class DescribeConsumerGroupTest {
    private static final String TOPIC_PREFIX = "test.topic.";
    private static final String GROUP_PREFIX = "test.group.";
    private static final List<List<String>> DESCRIBE_TYPE_OFFSETS = List.of(List.of(""), List.of("--offsets"), List.of("--offsets", "--verbose"));
    private static final List<List<String>> DESCRIBE_TYPE_MEMBERS = List.of(List.of("--members"), List.of("--members", "--verbose"));
    private static final List<List<String>> DESCRIBE_TYPE_STATE = List.of(List.of("--state"), List.of("--state", "--verbose"));
    private static final List<List<String>> DESCRIBE_TYPES = Stream.of(DESCRIBE_TYPE_OFFSETS, DESCRIBE_TYPE_MEMBERS, DESCRIBE_TYPE_STATE).flatMap(Collection::stream).toList();
    private ClusterInstance clusterInstance;

    @ClusterTest
    public void testDescribeNonExistingGroup(ClusterInstance clusterInstance) {
        String missingGroup = "missing.group";

        for (List<String> describeType : DESCRIBE_TYPES) {
            // note the group to be queried is a different (non-existing) group
            List<String> cgcArgs = new ArrayList<>(List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", missingGroup));
            cgcArgs.addAll(describeType);
            try (ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(cgcArgs.toArray(new String[0]))) {
                service.describeGroups();
                fail("Expected error was not detected for describe option '" + String.join(" ", describeType) + "'");
            } catch (ExecutionException ee) {
                assertInstanceOf(GroupIdNotFoundException.class, ee.getCause());
                assertEquals("Group " + missingGroup + " not found.", ee.getCause().getMessage());
            } catch (Exception e) {
                fail("Expected error was not detected for describe option '" + String.join(" ", describeType) + "'");
            }
        }
    }

    @ClusterTest
    public void testDescribeOffsetsOfNonExistingGroup(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        String missingGroup = "missing.group";
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic);

            // run one consumer in the group consuming from a single-partition topic
            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
                 // note the group to be queried is a different (non-existing) group
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", missingGroup})
            ) {
                service.collectGroupOffsets(missingGroup);
                fail("Expected the group '" + missingGroup + "' to throw GroupIdNotFoundException");
            } catch (ExecutionException ee) {
                assertInstanceOf(GroupIdNotFoundException.class, ee.getCause(),
                    "Expected the group '" + missingGroup + "' to throw GroupIdNotFoundException");
            }
        }
    }

    @ClusterTest
    public void testDescribeMembersOfNonExistingGroup(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        String missingGroup = "missing.group";
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic);
            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
                 // note the group to be queried is a different (non-existing) group
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", missingGroup})
            ) {
                service.collectGroupMembers(missingGroup);
                fail("Expected the group '" + missingGroup + "' to throw GroupIdNotFoundException");
            } catch (ExecutionException ee) {
                assertInstanceOf(GroupIdNotFoundException.class, ee.getCause(),
                    "Expected the group '" + missingGroup + "' to throw GroupIdNotFoundException");
            }
        }
    }

    @ClusterTest
    public void testDescribeStateOfNonExistingGroup(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        String missingGroup = "missing.group";
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic);
            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
                 // note the group to be queried is a different (non-existing) group
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", missingGroup})
            ) {
                service.collectGroupState(missingGroup);
                fail("Expected the group '" + missingGroup + "' to throw GroupIdNotFoundException");
            } catch (ExecutionException ee) {
                assertInstanceOf(GroupIdNotFoundException.class, ee.getCause(),
                        "Expected the group '" + missingGroup + "' to throw GroupIdNotFoundException");
            }
        }
    }

    @ClusterTest
    public void testDescribeGroupOffsets(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            clusterInstance.createTopic(topic, 1, (short) 1);
            sendRecords(topic, 0, 1);

            for (List<String> describeType : DESCRIBE_TYPE_OFFSETS) {
                String group = GROUP_PREFIX + groupProtocol.name() + "." + String.join("", describeType);
                List<String> cgcArgs = new ArrayList<>(List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group));
                cgcArgs.addAll(describeType);
                try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
                     ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(cgcArgs.toArray(new String[0]));
                     Admin admin = clusterInstance.admin()
                ) {
                    TestUtils.waitForCondition(() -> {
                        Entry<String, String> res = ToolsTestUtils.grabConsoleOutputAndError(describeGroups(service));
                        String[] lines = res.getKey().trim().split("\n");
                        if (lines.length != 2 && !res.getValue().isEmpty()) {
                            return false;
                        }
                        ConsumerGroupDescription consumerGroupDescription = admin.describeConsumerGroups(Set.of(group)).describedGroups().get(group).get();
                        MemberDescription memberDescription = consumerGroupDescription.members().iterator().next();

                        List<String> expectedValues;
                        if (describeType.contains("--verbose")) {
                            expectedValues = List.of(group, topic, "0", "-", "1", "1", "0", memberDescription.consumerId(),
                                memberDescription.host(), memberDescription.clientId());
                        } else {
                            expectedValues = List.of(group, topic, "0", "1", "1", "0", memberDescription.consumerId(),
                                memberDescription.host(), memberDescription.clientId());
                        }
                        return checkArgsHeaderOutput(cgcArgs, lines[0]) &&
                            Arrays.stream(lines[1].trim().split("\\s+")).toList().equals(expectedValues);
                    }, "Expected a data row and no error in describe results with describe type " + String.join(" ", describeType) + ".");
                }
            }
        }
    }

    @ClusterTest
    public void testDescribeGroupMembers(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            boolean isConsumer = groupProtocol.equals(GroupProtocol.CONSUMER);
            String topic1 = TOPIC_PREFIX + groupProtocol.name() + "1";
            String topic2 = TOPIC_PREFIX + groupProtocol.name() + "2";
            clusterInstance.createTopic(topic1, 2, (short) 1);
            clusterInstance.createTopic(topic2, 1, (short) 1);

            for (List<String> describeType : DESCRIBE_TYPE_MEMBERS) {
                String group = GROUP_PREFIX + groupProtocol.name() + "." + String.join("", describeType);
                List<String> cgcArgs = new ArrayList<>(List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group));
                cgcArgs.addAll(describeType);
                try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, Set.of(topic1, topic2), Map.of(), 1);
                     ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(cgcArgs.toArray(new String[0]));
                     Admin admin = clusterInstance.admin()
                ) {
                    TestUtils.waitForCondition(() -> {
                        Entry<String, String> res = ToolsTestUtils.grabConsoleOutputAndError(describeGroups(service));
                        String[] lines = res.getKey().trim().split("\n");
                        if (lines.length != 2 && !res.getValue().isEmpty()) {
                            return false;
                        }
                        ConsumerGroupDescription consumerGroupDescription = admin.describeConsumerGroups(Set.of(group)).describedGroups().get(group).get();
                        MemberDescription memberDescription = consumerGroupDescription.members().iterator().next();
                        String topicAssignment = topic1 + ":0,1;" + topic2 + ":0";

                        List<String> expectedValues;
                        if (describeType.contains("--verbose")) {
                            expectedValues = List.of(group, memberDescription.consumerId(), memberDescription.host(),
                                memberDescription.clientId(), "3", isConsumer ? memberDescription.memberEpoch().get().toString() : "-",
                                topicAssignment, isConsumer ? consumerGroupDescription.targetAssignmentEpoch().get().toString() : "-",
                                isConsumer ? topicAssignment : "-");
                        } else {
                            expectedValues = List.of(group, memberDescription.consumerId(), memberDescription.host(),
                                memberDescription.clientId(), "3");
                        }
                        return checkArgsHeaderOutput(cgcArgs, lines[0]) &&
                            Arrays.stream(lines[1].trim().split("\\s+")).toList().equals(expectedValues);
                    }, "Expected a data row and no error in describe results with describe type " + String.join(" ", describeType) + ".");
                }
            }
        }
    }

    @ClusterTest
    public void testDescribeGroupMemberWithMigration(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        String topic = TOPIC_PREFIX + "migration";
        String group = GROUP_PREFIX + "migration";
        String classicClientId = "classic";
        String consumerClientId = "consumer";
        clusterInstance.createTopic(topic, 2, (short) 1);
        List<String> cgcArgs = List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group, "--members", "--verbose");
        try (AutoCloseable classicConsumer = consumerGroupClosable(GroupProtocol.CLASSIC, group, topic, Map.of(ConsumerConfig.CLIENT_ID_CONFIG, classicClientId));
             Admin admin = clusterInstance.admin()
        ) {
            // Make sure the classic consumer is stable before starting another consumer.
            TestUtils.waitForCondition(
                () -> {
                    ConsumerGroupDescription consumerGroupDescription = admin.describeConsumerGroups(Set.of(group)).describedGroups().get(group).get();
                    MemberDescription memberDescription = consumerGroupDescription.members().iterator().next();
                    return !memberDescription.assignment().topicPartitions().isEmpty();
                },
                "Expected the classic consumer to join the group."
            );

            try (AutoCloseable consumerConsumer = consumerGroupClosable(GroupProtocol.CONSUMER, group, topic, Map.of(ConsumerConfig.CLIENT_ID_CONFIG, consumerClientId));
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(cgcArgs.toArray(new String[0]))
            ) {
                TestUtils.waitForCondition(() -> {
                    Entry<String, String> res = ToolsTestUtils.grabConsoleOutputAndError(describeGroups(service));
                    String[] lines = res.getKey().trim().split("\n");
                    if (lines.length != 3 && !res.getValue().isEmpty()) {
                        return false;
                    }

                    // We can't guarantee the order of the lines and final assignment on both members,
                    // so only checking UPGRADED field to make sure the tool can reflect member migration.
                    String[] header = lines[0].trim().split("\\s+");
                    assertEquals("UPGRADED", header[header.length - 1]);
                    List<String> line1 = Arrays.stream(lines[1].trim().split("\\s+")).toList();
                    List<String> line2 = Arrays.stream(lines[2].trim().split("\\s+")).toList();
                    if (line1.contains(classicClientId)) {
                        assertEquals("false", line1.get(line1.size() - 1));
                        assertEquals("true", line2.get(line2.size() - 1));
                    } else {
                        assertEquals("false", line2.get(line2.size() - 1));
                        assertEquals("true", line1.get(line1.size() - 1));
                    }
                    return true;
                }, "Expected a data row and no error in describe results with describe type \"--members --verbose\"");
            }
        }
    }

    @ClusterTest
    public void testDescribeGroupState(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            boolean isConsumer = groupProtocol.equals(GroupProtocol.CONSUMER);
            String topic = TOPIC_PREFIX + groupProtocol.name();
            clusterInstance.createTopic(topic, 1, (short) 1);

            for (List<String> describeType : DESCRIBE_TYPE_STATE) {
                String group = GROUP_PREFIX + groupProtocol.name() + "." + String.join("", describeType);
                List<String> cgcArgs = new ArrayList<>(List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group));
                cgcArgs.addAll(describeType);
                try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
                     ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(cgcArgs.toArray(new String[0]));
                     Admin admin = clusterInstance.admin()
                ) {
                    TestUtils.waitForCondition(() -> {
                        Entry<String, String> res = ToolsTestUtils.grabConsoleOutputAndError(describeGroups(service));
                        String[] lines = res.getKey().trim().split("\n");
                        if (lines.length != 2 && !res.getValue().isEmpty()) {
                            return false;
                        }
                        ConsumerGroupDescription consumerGroupDescription = admin.describeConsumerGroups(Set.of(group)).describedGroups().get(group).get();

                        List<String> expectedValues;
                        String coordinatorAddress = consumerGroupDescription.coordinator().host() + ":" + consumerGroupDescription.coordinator().port();
                        String coordinatorId = "(" + consumerGroupDescription.coordinator().idString() + ")";
                        if (describeType.contains("--verbose")) {
                            expectedValues = List.of(group, coordinatorAddress, coordinatorId, consumerGroupDescription.partitionAssignor(), GroupState.STABLE.toString(),
                                isConsumer ? consumerGroupDescription.groupEpoch().get().toString() : "-",
                                isConsumer ? consumerGroupDescription.targetAssignmentEpoch().get().toString() : "-", "1");
                        } else {
                            expectedValues = List.of(group, coordinatorAddress, coordinatorId, consumerGroupDescription.partitionAssignor(), GroupState.STABLE.toString(), "1");
                        }
                        return checkArgsHeaderOutput(cgcArgs, lines[0]) &&
                            Arrays.stream(lines[1].trim().split("\\s+")).toList().equals(expectedValues);
                    }, "Expected two data rows and no error in describe results with describe type " + String.join(" ", describeType) + ".");
                }
            }
        }
    }

    @ClusterTest
    public void testDescribeExistingGroups(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            createTopic(topic);

            // Create N single-threaded consumer groups from a single-partition topic
            List<AutoCloseable> protocolConsumerGroupExecutors = new ArrayList<>();
            try {
                List<String> groups = new ArrayList<>();
                for (List<String> describeType : DESCRIBE_TYPES) {
                    String group = GROUP_PREFIX + groupProtocol.name() + "." + String.join("", describeType);
                    groups.addAll(List.of("--group", group));
                    protocolConsumerGroupExecutors.add(consumerGroupClosable(groupProtocol, group, topic, Map.of()));
                }

                int expectedNumLines = DESCRIBE_TYPES.size() * 2;

                for (List<String> describeType : DESCRIBE_TYPES) {
                    List<String> cgcArgs = new ArrayList<>(List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe"));
                    cgcArgs.addAll(groups);
                    cgcArgs.addAll(describeType);
                    try (ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(cgcArgs.toArray(new String[0]))) {
                        TestUtils.waitForCondition(() -> {
                            Entry<String, String> res = ToolsTestUtils.grabConsoleOutputAndError(describeGroups(service));
                            long numLines = Arrays.stream(res.getKey().trim().split("\n")).filter(line -> !line.isEmpty()).count();
                            return (numLines == expectedNumLines) &&
                                res.getValue().isEmpty() &&
                                checkArgsHeaderOutput(cgcArgs, res.getKey().trim().split("\n")[0]);
                        }, "Expected a data row and no error in describe results with describe type " + String.join(" ", describeType) + ".");
                    }
                }
            } finally {
                for (AutoCloseable protocolConsumerGroupExecutor : protocolConsumerGroupExecutors) {
                    protocolConsumerGroupExecutor.close();
                }
            }
        }
    }

    @ClusterTest
    public void testDescribeAllExistingGroups(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            createTopic(topic);

            // Create N single-threaded consumer groups from a single-partition topic
            List<AutoCloseable> protocolConsumerGroupExecutors = new ArrayList<>();
            List<String> groups = new ArrayList<>();
            try {
                for (List<String> describeType : DESCRIBE_TYPES) {
                    String group = GROUP_PREFIX + groupProtocol.name() + "." + String.join("", describeType);
                    groups.add(group);
                    protocolConsumerGroupExecutors.add(consumerGroupClosable(groupProtocol, group, topic, Map.of()));
                }
                int expectedNumLines = DESCRIBE_TYPES.size() * 2;
                for (List<String> describeType : DESCRIBE_TYPES) {
                    List<String> cgcArgs = new ArrayList<>(List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--all-groups"));
                    cgcArgs.addAll(describeType);
                    try (ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(cgcArgs.toArray(new String[0]))) {
                        TestUtils.waitForCondition(() -> {
                            Entry<String, String> res = ToolsTestUtils.grabConsoleOutputAndError(describeGroups(service));
                            long numLines = Arrays.stream(res.getKey().trim().split("\n")).filter(line -> !line.isEmpty()).count();
                            return (numLines == expectedNumLines) &&
                                res.getValue().isEmpty() &&
                                checkArgsHeaderOutput(cgcArgs, res.getKey().trim().split("\n")[0]);
                        }, "Expected a data row and no error in describe results with describe type " + String.join(" ", describeType) + ".");
                    }
                }
            } finally {
                for (AutoCloseable protocolConsumerGroupExecutor : protocolConsumerGroupExecutors) {
                    protocolConsumerGroupExecutor.close();
                }
                // remove previous consumer groups, so we can have a clean cluster for next consumer group protocol test.
                deleteConsumerGroups(groups);
                deleteTopic(topic);
            }
        }
    }

    @ClusterTest
    public void testDescribeOffsetsOfExistingGroup(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic);

            // run one consumer in the group consuming from a single-partition topic
            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group})
            ) {
                TestUtils.waitForCondition(() -> {
                    Entry<Optional<GroupState>, Optional<Collection<PartitionAssignmentState>>> groupOffsets = service.collectGroupOffsets(group);
                    Optional<GroupState> state = groupOffsets.getKey();
                    Optional<Collection<PartitionAssignmentState>> assignments = groupOffsets.getValue();

                    Predicate<PartitionAssignmentState> isGrp = s -> Objects.equals(s.group(), group);

                    boolean res = state.map(s -> s.equals(GroupState.STABLE)).orElse(false) &&
                            assignments.isPresent() &&
                            assignments.get().stream().filter(isGrp).count() == 1;

                    if (!res)
                        return false;

                    Optional<PartitionAssignmentState> maybePartitionState = assignments.get().stream().filter(isGrp).findFirst();
                    if (maybePartitionState.isEmpty())
                        return false;

                    PartitionAssignmentState partitionState = maybePartitionState.get();

                    return !partitionState.consumerId().map(s0 -> s0.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false) &&
                            !partitionState.clientId().map(s0 -> s0.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false) &&
                            !partitionState.host().map(h -> h.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false);
                }, "Expected a 'Stable' group status, rows and valid values for consumer id / client id / host columns in describe results for group " + group + ".");
            }
        }
    }

    @ClusterTest
    public void testDescribeMembersOfExistingGroup(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic);

            // run one consumer in the group consuming from a single-partition topic
            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group});
                 Admin admin = Admin.create(Map.of(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers()))
            ) {
                TestUtils.waitForCondition(() -> {
                    ConsumerGroupDescription consumerGroupDescription = admin.describeConsumerGroups(Set.of(group)).describedGroups().get(group).get();
                    return consumerGroupDescription.members().size() == 1 && consumerGroupDescription.members().iterator().next().assignment().topicPartitions().size() == 1;
                }, "Expected a 'Stable' group status, rows and valid member information for group " + group + ".");

                Entry<Optional<GroupState>, Optional<Collection<MemberAssignmentState>>> res = service.collectGroupMembers(group);

                assertTrue(res.getValue().isPresent());
                assertTrue(res.getValue().get().size() == 1 && res.getValue().get().iterator().next().assignment().size() == 1,
                        "Expected a topic partition assigned to the single group member for group " + group);
            }
        }
    }

    @ClusterTest
    public void testDescribeStateOfExistingGroup(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic);

            // run one consumer in the group consuming from a single-partition topic
            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of(ConsumerConfig.GROUP_REMOTE_ASSIGNOR_CONFIG, groupProtocol == GroupProtocol.CONSUMER ? "range" : ""));
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group})
            ) {
                TestUtils.waitForCondition(() -> {
                    GroupInformation state = service.collectGroupState(group);
                    return Objects.equals(state.groupState(), GroupState.STABLE) &&
                            state.numMembers() == 1 &&
                            state.coordinator() != null &&
                            clusterInstance.brokerIds().contains(state.coordinator().id());
                }, "Expected a 'Stable' group status, with one member for group " + group + ".");
            }
        }
    }

    @ClusterTest
    public void testDescribeStateOfExistingGroupWithNonDefaultAssignor(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic);

            // run one consumer in the group consuming from a single-partition topic
            AutoCloseable protocolConsumerGroupExecutor = null;
            try {
                String expectedName;
                if (groupProtocol.equals(GroupProtocol.CONSUMER)) {
                    protocolConsumerGroupExecutor = consumerGroupClosable(GroupProtocol.CONSUMER, group, topic, Map.of(ConsumerConfig.GROUP_REMOTE_ASSIGNOR_CONFIG, "range"));
                    expectedName = RangeAssignor.RANGE_ASSIGNOR_NAME;
                } else {
                    protocolConsumerGroupExecutor = consumerGroupClosable(GroupProtocol.CLASSIC, group, topic, Map.of(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RoundRobinAssignor.class.getName()));
                    expectedName = RoundRobinAssignor.ROUNDROBIN_ASSIGNOR_NAME;
                }

                try (ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group})) {
                    TestUtils.waitForCondition(() -> {
                        GroupInformation state = service.collectGroupState(group);
                        return Objects.equals(state.groupState(), GroupState.STABLE) &&
                                state.numMembers() == 1 &&
                                Objects.equals(state.assignmentStrategy(), expectedName) &&
                                state.coordinator() != null &&
                                clusterInstance.brokerIds().contains(state.coordinator().id());
                    }, "Expected a 'Stable' group status, with one member and " + expectedName + " assignment strategy for group " + group + ".");
                }
            } finally {
                if (protocolConsumerGroupExecutor != null) {
                    protocolConsumerGroupExecutor.close();
                }
            }
        }
    }

    @ClusterTest
    public void testDescribeExistingGroupWithNoMembers(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            createTopic(topic);

            for (List<String> describeType : DESCRIBE_TYPES) {
                String group = GROUP_PREFIX + groupProtocol.name() + String.join("", describeType);
                List<String> cgcArgs = new ArrayList<>(List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group));
                cgcArgs.addAll(describeType);
                // run one consumer in the group consuming from a single-partition topic
                try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
                     ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(cgcArgs.toArray(new String[0]))
                ) {
                    TestUtils.waitForCondition(() -> {
                        Entry<String, String> res = ToolsTestUtils.grabConsoleOutputAndError(describeGroups(service));
                        return res.getKey().trim().split("\n").length == 2 &&
                            res.getValue().isEmpty() &&
                            checkArgsHeaderOutput(cgcArgs, res.getKey().trim().split("\n")[0]);
                    }, "Expected describe group results with one data row for describe type '" + String.join(" ", describeType) + "'");
                    
                    protocolConsumerGroupExecutor.close();
                    TestUtils.waitForCondition(
                            () -> ToolsTestUtils.grabConsoleError(describeGroups(service)).contains("Consumer group '" + group + "' has no active members."),
                            "Expected no active member in describe group results with describe type " + String.join(" ", describeType));
                }
            }
        }
    }

    @ClusterTest
    public void testDescribeOffsetsOfExistingGroupWithNoMembers(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic);

            // run one consumer in the group consuming from a single-partition topic
            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group})
            ) {
                TestUtils.waitForCondition(() -> {
                    Entry<Optional<GroupState>, Optional<Collection<PartitionAssignmentState>>> res = service.collectGroupOffsets(group);
                    return res.getKey().map(s -> s.equals(GroupState.STABLE)).orElse(false)
                            && res.getValue().map(c -> c.stream().anyMatch(assignment -> Objects.equals(assignment.group(), group) && assignment.offset().isPresent())).orElse(false);
                }, "Expected the group to initially become stable, and to find group in assignments after initial offset commit.");

                // stop the consumer so the group has no active member anymore
                protocolConsumerGroupExecutor.close();

                TestUtils.waitForCondition(() -> {
                    Entry<Optional<GroupState>, Optional<Collection<PartitionAssignmentState>>> offsets = service.collectGroupOffsets(group);
                    Optional<GroupState> state = offsets.getKey();
                    Optional<Collection<PartitionAssignmentState>> assignments = offsets.getValue();
                    List<PartitionAssignmentState> testGroupAssignments = assignments.get().stream().filter(a -> Objects.equals(a.group(), group)).toList();
                    PartitionAssignmentState assignment = testGroupAssignments.get(0);
                    return state.map(s -> s.equals(GroupState.EMPTY)).orElse(false) &&
                            testGroupAssignments.size() == 1 &&
                            assignment.consumerId().map(c -> c.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false) && // the member should be gone
                            assignment.clientId().map(c -> c.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false) &&
                            assignment.host().map(c -> c.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false);
                }, "failed to collect group offsets");
            }
        }
    }

    @ClusterTest
    public void testDescribeMembersOfExistingGroupWithNoMembers(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic);

            // run one consumer in the group consuming from a single-partition topic
            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group})
            ) {
                TestUtils.waitForCondition(() -> {
                    Entry<Optional<GroupState>, Optional<Collection<MemberAssignmentState>>> res = service.collectGroupMembers(group);
                    return res.getKey().map(s -> s.equals(GroupState.STABLE)).orElse(false)
                            && res.getValue().map(c -> c.stream().anyMatch(m -> Objects.equals(m.group(), group))).orElse(false);
                }, "Expected the group to initially become stable, and to find group in assignments after initial offset commit.");

                // stop the consumer so the group has no active member anymore
                protocolConsumerGroupExecutor.close();

                TestUtils.waitForCondition(() -> {
                    Entry<Optional<GroupState>, Optional<Collection<MemberAssignmentState>>> res = service.collectGroupMembers(group);
                    return res.getKey().map(s -> s.equals(GroupState.EMPTY)).orElse(false) && res.getValue().isPresent() && res.getValue().get().isEmpty();
                }, "Expected no member in describe group members results for group '" + group + "'");
            }
        }
    }

    @ClusterTest
    public void testDescribeStateOfExistingGroupWithNoMembers(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic);

            // run one consumer in the group consuming from a single-partition topic
            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group})
            ) {
                TestUtils.waitForCondition(() -> {
                    GroupInformation state = service.collectGroupState(group);
                    return Objects.equals(state.groupState(), GroupState.STABLE) &&
                            state.numMembers() == 1 &&
                            state.coordinator() != null &&
                            clusterInstance.brokerIds().contains(state.coordinator().id());
                }, "Expected the group to initially become stable, and have a single member.");

                // stop the consumer so the group has no active member anymore
                protocolConsumerGroupExecutor.close();

                TestUtils.waitForCondition(() -> {
                    GroupInformation state = service.collectGroupState(group);
                    return Objects.equals(state.groupState(), GroupState.EMPTY) && state.numMembers() == 0;
                }, "Expected the group to become empty after the only member leaving.");
            }
        }
    }

    @ClusterTest
    public void testDescribeWithConsumersWithoutAssignedPartitions(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            createTopic(topic);

            for (List<String> describeType : DESCRIBE_TYPES) {
                String group = GROUP_PREFIX + groupProtocol.name() + String.join("", describeType);
                List<String> cgcArgs = new ArrayList<>(List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group));
                cgcArgs.addAll(describeType);
                // run two consumers in the group consuming from a single-partition topic
                try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of(), 2);
                     ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(cgcArgs.toArray(new String[0]))
                ) {
                    TestUtils.waitForCondition(() -> {
                        Entry<String, String> res = ToolsTestUtils.grabConsoleOutputAndError(describeGroups(service));
                        int expectedNumRows = DESCRIBE_TYPE_MEMBERS.contains(describeType) ? 3 : 2;
                        return res.getValue().isEmpty() &&
                            res.getKey().trim().split("\n").length == expectedNumRows &&
                            checkArgsHeaderOutput(cgcArgs, res.getKey().trim().split("\n")[0]);
                    }, "Expected a single data row in describe group result with describe type '" + String.join(" ", describeType) + "'");
                }
            }
        }
    }

    @ClusterTest
    public void testDescribeOffsetsWithConsumersWithoutAssignedPartitions(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic);

            // run two consumers in the group consuming from a single-partition topic
            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of(), 2);
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group})
            ) {
                TestUtils.waitForCondition(() -> {
                    Entry<Optional<GroupState>, Optional<Collection<PartitionAssignmentState>>> res = service.collectGroupOffsets(group);
                    return res.getKey().map(s -> s.equals(GroupState.STABLE)).isPresent() &&
                            res.getValue().isPresent() &&
                            res.getValue().get().stream().filter(s -> Objects.equals(s.group(), group)).count() == 1 &&
                            res.getValue().get().stream().filter(x -> Objects.equals(x.group(), group) && x.partition().isPresent()).count() == 1;
                }, "Expected rows for consumers with no assigned partitions in describe group results");
            }
        }
    }

    @ClusterTest
    public void testDescribeMembersWithConsumersWithoutAssignedPartitions(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic);

            // run two consumers in the group consuming from a single-partition topic
            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of(), 2);
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group})
            ) {
                TestUtils.waitForCondition(() -> {
                    Entry<Optional<GroupState>, Optional<Collection<MemberAssignmentState>>> res = service.collectGroupMembers(group);
                    return res.getKey().map(s -> s.equals(GroupState.STABLE)).orElse(false) &&
                            res.getValue().isPresent() &&
                            res.getValue().get().stream().filter(s -> Objects.equals(s.group(), group)).count() == 2 &&
                            res.getValue().get().stream().filter(x -> Objects.equals(x.group(), group) && x.numPartitions() == 1).count() == 1 &&
                            res.getValue().get().stream().filter(x -> Objects.equals(x.group(), group) && x.numPartitions() == 0).count() == 1 &&
                            res.getValue().get().stream().anyMatch(s -> !s.assignment().isEmpty());
                }, "Expected rows for consumers with no assigned partitions in describe group results");

                Entry<Optional<GroupState>, Optional<Collection<MemberAssignmentState>>> res = service.collectGroupMembers(group);
                assertTrue(res.getKey().map(s -> s.equals(GroupState.STABLE)).orElse(false)
                                && res.getValue().map(c -> c.stream().anyMatch(s -> !s.assignment().isEmpty())).orElse(false),
                        "Expected additional columns in verbose version of describe members");
            }
        }
    }

    @ClusterTest
    public void testDescribeStateWithConsumersWithoutAssignedPartitions(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic);

            // run two consumers in the group consuming from a single-partition topic
            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of(), 2);
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group})
            ) {
                TestUtils.waitForCondition(() -> {
                    GroupInformation state = service.collectGroupState(group);
                    return Objects.equals(state.groupState(), GroupState.STABLE) && state.numMembers() == 2;
                }, "Expected two consumers in describe group results");
            }
        }
    }

    @ClusterTest
    public void testDescribeWithMultiPartitionTopicAndMultipleConsumers(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            createTopic(topic, 2);

            for (List<String> describeType : DESCRIBE_TYPES) {
                String group = GROUP_PREFIX + groupProtocol.name() + String.join("", describeType);
                List<String> cgcArgs = new ArrayList<>(List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group));
                cgcArgs.addAll(describeType);
                // run two consumers in the group consuming from a two-partition topic
                try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of(), 2);
                     ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(cgcArgs.toArray(new String[0]))
                ) {
                    TestUtils.waitForCondition(() -> {
                        Entry<String, String> res = ToolsTestUtils.grabConsoleOutputAndError(describeGroups(service));
                        int expectedNumRows = DESCRIBE_TYPE_STATE.contains(describeType) ? 2 : 3;
                        return res.getValue().isEmpty() &&
                            res.getKey().trim().split("\n").length == expectedNumRows &&
                            checkArgsHeaderOutput(cgcArgs, res.getKey().trim().split("\n")[0]);
                    }, "Expected a single data row in describe group result with describe type '" + String.join(" ", describeType) + "'");
                }
            }
        }
    }

    @ClusterTest
    public void testDescribeOffsetsWithMultiPartitionTopicAndMultipleConsumers(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic, 2);

            // run two consumers in the group consuming from a two-partition topic
            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of(), 2);
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group})
            ) {
                TestUtils.waitForCondition(() -> {
                    Entry<Optional<GroupState>, Optional<Collection<PartitionAssignmentState>>> res = service.collectGroupOffsets(group);
                    return res.getKey().map(s -> s.equals(GroupState.STABLE)).orElse(false) &&
                            res.getValue().isPresent() &&
                            res.getValue().get().stream().filter(s -> Objects.equals(s.group(), group)).count() == 2 &&
                            res.getValue().get().stream().filter(x -> Objects.equals(x.group(), group) && x.partition().isPresent()).count() == 2 &&
                            res.getValue().get().stream().noneMatch(x -> Objects.equals(x.group(), group) && x.partition().isEmpty());
                }, "Expected two rows (one row per consumer) in describe group results.");
            }
        }
    }

    @ClusterTest
    public void testDescribeMembersWithMultiPartitionTopicAndMultipleConsumers(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic, 2);

            // run two consumers in the group consuming from a two-partition topic
            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of(), 2);
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group})
            ) {
                TestUtils.waitForCondition(() -> {
                    Entry<Optional<GroupState>, Optional<Collection<MemberAssignmentState>>> res = service.collectGroupMembers(group);
                    return res.getKey().map(s -> s.equals(GroupState.STABLE)).orElse(false) &&
                            res.getValue().isPresent() &&
                            res.getValue().get().stream().filter(s -> Objects.equals(s.group(), group)).count() == 2 &&
                            res.getValue().get().stream().filter(x -> Objects.equals(x.group(), group) && x.numPartitions() == 1).count() == 2 &&
                            res.getValue().get().stream().noneMatch(x -> Objects.equals(x.group(), group) && x.numPartitions() == 0);
                }, "Expected two rows (one row per consumer) in describe group members results.");

                Entry<Optional<GroupState>, Optional<Collection<MemberAssignmentState>>> res = service.collectGroupMembers(group);
                assertTrue(res.getKey().map(s -> s.equals(GroupState.STABLE)).orElse(false) && res.getValue().map(s -> s.stream().filter(x -> x.assignment().isEmpty()).count()).orElse(0L) == 0,
                        "Expected additional columns in verbose version of describe members");
            }
        }
    }

    @ClusterTest
    public void testDescribeStateWithMultiPartitionTopicAndMultipleConsumers(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic, 2);

            // run two consumers in the group consuming from a two-partition topic
            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of(), 2);
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group})
            ) {
                TestUtils.waitForCondition(() -> {
                    GroupInformation state = service.collectGroupState(group);
                    return Objects.equals(state.groupState(), GroupState.STABLE) && Objects.equals(state.group(), group) && state.numMembers() == 2;
                }, "Expected a stable group with two members in describe group state result.");
            }
        }
    }

    @ClusterTest
    public void testDescribeSimpleConsumerGroup(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        // Ensure that the offsets of consumers which don't use group management are still displayed
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic, 2);

            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(GroupProtocol.CLASSIC, group, Set.of(new TopicPartition(topic, 0), new TopicPartition(topic, 1)), Map.of());
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group})
            ) {
                TestUtils.waitForCondition(() -> {
                    Entry<Optional<GroupState>, Optional<Collection<PartitionAssignmentState>>> res = service.collectGroupOffsets(group);
                    return res.getKey().map(s -> s.equals(GroupState.EMPTY)).orElse(false)
                            && res.getValue().isPresent() && res.getValue().get().stream().filter(s -> Objects.equals(s.group(), group)).count() == 2;
                }, "Expected a stable group with two members in describe group state result.");
            }
        }
    }

    @ClusterTest
    public void testDescribeGroupWithShortInitializationTimeout(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            createTopic(topic);

            // Let creation of the offsets topic happen during group initialization to ensure that initialization doesn't
            // complete before the timeout expires
            List<String> describeType = DESCRIBE_TYPES.get(RANDOM.nextInt(DESCRIBE_TYPES.size()));
            String group = GROUP_PREFIX + groupProtocol.name() + String.join("", describeType);

            // set the group initialization timeout too low for the group to stabilize
            List<String> cgcArgs = new ArrayList<>(List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--timeout", "1", "--group", group));
            cgcArgs.addAll(describeType);

            // run one consumer in the group consuming from a single-partition topic
            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(cgcArgs.toArray(new String[0]))
            ) {
                ExecutionException e = assertThrows(ExecutionException.class, service::describeGroups);
                assertInstanceOf(TimeoutException.class, e.getCause());
            }
        }
    }

    @ClusterTest
    public void testDescribeGroupOffsetsWithShortInitializationTimeout(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic);

            // Let creation of the offsets topic happen during group initialization to ensure that initialization doesn't
            // complete before the timeout expires

            // run one consumer in the group consuming from a single-partition topic
            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
                 // set the group initialization timeout too low for the group to stabilize
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group, "--timeout", "1"})
            ) {
                Throwable e = assertThrows(ExecutionException.class, () -> service.collectGroupOffsets(group));
                assertEquals(TimeoutException.class, e.getCause().getClass());
            }
        }
    }

    @ClusterTest
    public void testDescribeGroupMembersWithShortInitializationTimeout(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic);

            // Let creation of the offsets topic happen during group initialization to ensure that initialization doesn't
            // complete before the timeout expires

            // run one consumer in the group consuming from a single-partition topic
            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
                 // set the group initialization timeout too low for the group to stabilize
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group, "--timeout", "1"})
            ) {
                Throwable e = assertThrows(ExecutionException.class, () -> service.collectGroupMembers(group));
                assertEquals(TimeoutException.class, e.getCause().getClass());
                e = assertThrows(ExecutionException.class, () -> service.collectGroupMembers(group));
                assertEquals(TimeoutException.class, e.getCause().getClass());
            }
        }
    }

    @ClusterTest
    public void testDescribeGroupStateWithShortInitializationTimeout(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic);

            // Let creation of the offsets topic happen during group initialization to ensure that initialization doesn't
            // complete before the timeout expires

            // run one consumer in the group consuming from a single-partition topic
            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
                 // set the group initialization timeout too low for the group to stabilize
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group, "--timeout", "1"})
            ) {
                Throwable e = assertThrows(ExecutionException.class, () -> service.collectGroupState(group));
                assertEquals(TimeoutException.class, e.getCause().getClass());
            }
        }
    }

    @ClusterTest
    public void testDescribeNonOffsetCommitGroup(ClusterInstance clusterInstance) throws Exception {
        this.clusterInstance = clusterInstance;
        for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
            String topic = TOPIC_PREFIX + groupProtocol.name();
            String group = GROUP_PREFIX + groupProtocol.name();
            createTopic(topic);

            // run one consumer in the group consuming from a single-partition topic
            try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"));
                 ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group})
            ) {
                TestUtils.waitForCondition(() -> {
                    Entry<Optional<GroupState>, Optional<Collection<PartitionAssignmentState>>> groupOffsets = service.collectGroupOffsets(group);

                    Predicate<PartitionAssignmentState> isGrp = s -> Objects.equals(s.group(), group);

                    boolean res = groupOffsets.getKey().map(s -> s.equals(GroupState.STABLE)).orElse(false) &&
                            groupOffsets.getValue().isPresent() &&
                            groupOffsets.getValue().get().stream().filter(isGrp).count() == 1;

                    if (!res)
                        return false;

                    Optional<PartitionAssignmentState> maybeAssignmentState = groupOffsets.getValue().get().stream().filter(isGrp).findFirst();
                    if (maybeAssignmentState.isEmpty())
                        return false;

                    PartitionAssignmentState assignmentState = maybeAssignmentState.get();

                    return assignmentState.consumerId().map(c -> !c.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false) &&
                            assignmentState.clientId().map(c -> !c.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false) &&
                            assignmentState.host().map(h -> !h.trim().equals(ConsumerGroupCommand.MISSING_COLUMN_VALUE)).orElse(false);
                }, "Expected a 'Stable' group status, rows and valid values for consumer id / client id / host columns in describe results for non-offset-committing group " + group + ".");
            }
        }
    }

    /**
     * The config `OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG` needs to be set to a value greater than 1 to ensure the
     * normal invocation of APIs such as `FIND_COORDINATOR` when a broker has shutdown
     */
    @Timeout(60)
    @ClusterTest(brokers = 3, serverProperties = {@ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "2")})
    public void testDescribeConsumerGroupWithoutLeaders(ClusterInstance clusterInstance) throws Exception {
        int brokerNum = 3;
        this.clusterInstance = clusterInstance;

        // define topic and group, then send 5 records to each partition
        String topic = TOPIC_PREFIX + UUID.randomUUID();
        String group = GROUP_PREFIX + UUID.randomUUID();
        clusterInstance.createTopic(topic, brokerNum, (short) 1);
        for (int i = 0; i < brokerNum; i++) {
            sendRecords(topic, i, 5);
        }

        // append the command
        List<String> cgcArgs = List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group, "--all-topics");

        try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(GroupProtocol.CLASSIC, group, topic, Map.of());
             ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(cgcArgs.toArray(new String[0]));
             Admin admin = clusterInstance.admin()
        ) {
            // shutdown the target broker
            int noneLeaderPartition = 2;
            int shutdownBrokerId = clusterInstance.getLeaderBrokerId(new TopicPartition(topic, noneLeaderPartition));
            clusterInstance.shutdownBroker(shutdownBrokerId);

            TestUtils.waitForCondition(() -> {
                Entry<String, String> res = ToolsTestUtils.grabConsoleOutputAndError(describeGroups(service));
                String[] lines = res.getKey().trim().split("\n");
                if (lines.length != 4 || !res.getValue().isEmpty()) {
                    return false;
                }

                // get the client data, such as `consumerId,host,clientId`, to append the expected output
                ConsumerGroupDescription consumerGroupDescription = admin.describeConsumerGroups(Set.of(group)).describedGroups().get(group).get();
                MemberDescription memberDescription = consumerGroupDescription.members().iterator().next();
                String consumerId = memberDescription.consumerId();
                String host = memberDescription.host();
                String clientId = memberDescription.clientId();

                // the expected output
                List<String> partition0content = List.of(group, topic, "0", "5", "5", "0", consumerId, host, clientId);
                List<String> partition1content = List.of(group, topic, "1", "5", "5", "0", consumerId, host, clientId);
                List<String> partition2content = List.of(group, topic, "2", "-", "-", "-", consumerId, host, clientId);

                return checkArgsHeaderOutput(cgcArgs, lines[0])
                        && Arrays.stream(lines[1].trim().split("\\s+")).toList().equals(partition0content)
                        && Arrays.stream(lines[2].trim().split("\\s+")).toList().equals(partition1content)
                        && Arrays.stream(lines[3].trim().split("\\s+")).toList().equals(partition2content);
            }, "Expected 3 data rows excluding the header and no error in describe groups when a broker shutdown.");
        }
    }


    @Test
    public void testDescribeWithUnrecognizedNewConsumerOption() {
        String group = GROUP_PREFIX +  "unrecognized";
        String[] cgcArgs = new String[]{"--new-consumer", "--bootstrap-server", "localhost:9092", "--describe", "--group", group};
        assertThrows(joptsimple.OptionException.class, () -> ConsumerGroupCommandOptions.fromArgs(cgcArgs));
    }

    @Test
    public void testDescribeWithMultipleSubActions() {
        String group = GROUP_PREFIX + "multiple.sub.actions";
        AtomicInteger exitStatus = new AtomicInteger(0);
        AtomicReference<String> exitMessage = new AtomicReference<>("");
        Exit.setExitProcedure((status, err) -> {
            exitStatus.set(status);
            exitMessage.set(err);
            throw new RuntimeException();
        });
        String[] cgcArgs = new String[]{"--bootstrap-server", "localhost:9092", "--describe", "--group", group, "--members", "--state"};
        try {
            assertThrows(RuntimeException.class, () -> ConsumerGroupCommand.main(cgcArgs));
        } finally {
            Exit.resetExitProcedure();
        }
        assertEquals(1, exitStatus.get());
        assertTrue(exitMessage.get().contains("Option [describe] takes at most one of these options"));
    }

    @Test
    public void testDescribeWithStateValue() {
        AtomicInteger exitStatus = new AtomicInteger(0);
        AtomicReference<String> exitMessage = new AtomicReference<>("");
        Exit.setExitProcedure((status, err) -> {
            exitStatus.set(status);
            exitMessage.set(err);
            throw new RuntimeException();
        });
        String[] cgcArgs = new String[]{"--bootstrap-server", "localhost:9092", "--describe", "--all-groups", "--state", "Stable"};
        try {
            assertThrows(RuntimeException.class, () -> ConsumerGroupCommand.main(cgcArgs));
        } finally {
            Exit.resetExitProcedure();
        }
        assertEquals(1, exitStatus.get());
        assertTrue(exitMessage.get().contains("Option [describe] does not take a value for [state]"));
    }

    @Test
    public void testPrintVersion() {
        ToolsTestUtils.MockExitProcedure exitProcedure = new ToolsTestUtils.MockExitProcedure();
        Exit.setExitProcedure(exitProcedure);
        try {
            String out = ToolsTestUtils.captureStandardOut(() -> ConsumerGroupCommandOptions.fromArgs(new String[]{"--version"}));
            assertEquals(0, exitProcedure.statusCode());
            assertEquals(AppInfoParser.getVersion(), out);
        } finally {
            Exit.resetExitProcedure();
        }
    }

    private static ConsumerGroupCommand.ConsumerGroupService consumerGroupService(String[] args) {
        return new ConsumerGroupCommand.ConsumerGroupService(
                ConsumerGroupCommandOptions.fromArgs(args),
                Map.of(AdminClientConfig.RETRIES_CONFIG, Integer.toString(Integer.MAX_VALUE))
        );
    }

    private void createTopic(String topic) {
        createTopic(topic, 1);
    }

    private void createTopic(String topic, int numPartitions) {
        try (Admin admin = Admin.create(Map.of(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers()))) {
            Assertions.assertDoesNotThrow(() -> admin.createTopics(List.of(new NewTopic(topic, numPartitions, (short) 1))).topicId(topic).get());
        }
    }

    private void deleteConsumerGroups(Collection<String> groupIds) {
        try (Admin admin = Admin.create(Map.of(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers()))) {
            Assertions.assertDoesNotThrow(() -> admin.deleteConsumerGroups(groupIds).all().get());
        }
    }

    private void deleteTopic(String topic) {
        try (Admin admin = Admin.create(Map.of(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers()))) {
            Assertions.assertDoesNotThrow(() -> admin.deleteTopics(List.of(topic)).topicNameValues().get(topic).get());
        }
    }

    private AutoCloseable consumerGroupClosable(GroupProtocol protocol, String groupId, Set<TopicPartition> topicPartitions, Map<String, Object> customConfigs) {
        Map<String, Object> configs = composeConfigs(
                groupId,
                protocol.name,
                customConfigs
        );
        return ConsumerGroupCommandTestUtils.buildConsumers(
                1,
                topicPartitions,
                () -> new KafkaConsumer<String, String>(configs)
        );
    }

    private AutoCloseable consumerGroupClosable(GroupProtocol protocol, String groupId, String topicName, Map<String, Object> customConfigs) {
        return consumerGroupClosable(protocol, groupId, topicName, customConfigs, 1);
    }

    private AutoCloseable consumerGroupClosable(GroupProtocol protocol, String groupId, String topicName, Map<String, Object> customConfigs, int numConsumers) {
        return consumerGroupClosable(protocol, groupId, Set.of(topicName), customConfigs, numConsumers);
    }

    private AutoCloseable consumerGroupClosable(
        GroupProtocol protocol,
        String groupId,
        Set<String> topicNames,
        Map<String, Object> customConfigs,
        int numConsumers
    ) {
        Map<String, Object> configs = composeConfigs(
                groupId,
                protocol.name,
                customConfigs
        );
        return ConsumerGroupCommandTestUtils.buildConsumers(
                numConsumers,
                true,
                () -> new KafkaConsumer<String, String>(configs),
                consumer -> consumer.subscribe(topicNames)
        );
    }

    private Map<String, Object> composeConfigs(String groupId, String groupProtocol, Map<String, Object> customConfigs) {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers());
        configs.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        configs.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol);

        configs.putAll(customConfigs);
        return configs;
    }

    private Runnable describeGroups(ConsumerGroupCommand.ConsumerGroupService service) {
        return () -> Assertions.assertDoesNotThrow(service::describeGroups);
    }

    private boolean checkArgsHeaderOutput(List<String> args, String output) {
        if (!output.contains("GROUP")) {
            return false;
        }

        if (args.contains("--members")) {
            return checkMembersArgsHeaderOutput(output, args.contains("--verbose"));
        }

        if (args.contains("--state")) {
            return checkStateArgsHeaderOutput(output, args.contains("--verbose"));
        }

        // --offsets or no arguments
        return checkOffsetsArgsHeaderOutput(output, args.contains("--verbose"));
    }

    private boolean checkOffsetsArgsHeaderOutput(String output, boolean verbose) {
        List<String> expectedKeys = verbose ?
            List.of("GROUP", "TOPIC", "PARTITION", "LEADER-EPOCH", "CURRENT-OFFSET", "LOG-END-OFFSET", "LAG", "CONSUMER-ID", "HOST", "CLIENT-ID") :
            List.of("GROUP", "TOPIC", "PARTITION", "CURRENT-OFFSET", "LOG-END-OFFSET", "LAG", "CONSUMER-ID", "HOST", "CLIENT-ID");
        return Arrays.stream(output.trim().split("\\s+")).toList().equals(expectedKeys);
    }

    private boolean checkStateArgsOutput(String output) {
        return output.contains("COORDINATOR (ID)") && output.contains("ASSIGNMENT-STRATEGY") && output.contains("STATE") && output.contains("#MEMBERS");
    }

    private boolean checkMembersArgsHeaderOutput(String output, boolean verbose) {
        List<String> expectedKeys = verbose ?
            List.of("GROUP", "CONSUMER-ID", "HOST", "CLIENT-ID", "#PARTITIONS", "CURRENT-EPOCH", "CURRENT-ASSIGNMENT", "TARGET-EPOCH", "TARGET-ASSIGNMENT") :
            List.of("GROUP", "CONSUMER-ID", "HOST", "CLIENT-ID", "#PARTITIONS");
        return Arrays.stream(output.trim().split("\\s+")).toList().equals(expectedKeys);
    }

    private boolean checkStateArgsHeaderOutput(String output, boolean verbose) {
        List<String> expectedKeys = verbose ?
            List.of("GROUP", "COORDINATOR", "(ID)", "ASSIGNMENT-STRATEGY", "STATE", "GROUP-EPOCH", "TARGET-ASSIGNMENT-EPOCH", "#MEMBERS") :
            List.of("GROUP", "COORDINATOR", "(ID)", "ASSIGNMENT-STRATEGY", "STATE", "#MEMBERS");
        return Arrays.stream(output.trim().split("\\s+")).toList().equals(expectedKeys);
    }

    private void sendRecords(String topic, int partition, int recordsCount) {
        try (KafkaProducer<String, String> producer = new KafkaProducer<>(Map.of(
            ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers(),
            ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName(),
            ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()
        ))) {
            IntStream.range(0, recordsCount).forEach(i ->
                producer.send(new org.apache.kafka.clients.producer.ProducerRecord<>(topic, partition, Integer.toString(i), Integer.toString(i))));
            producer.flush();
        }
    }
}
