Repository: apache/kafka
File: tools/src/test/java/org/apache/kafka/tools/consumer/group/DescribeConsumerGroupTest.java (Java)

Code before:
                            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
     */
    @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 = new ArrayList<>(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

Review comment (suggestion):
```suggestion
        List<String> cgcArgs = List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group, "--all-topics");
```

Code after:
                    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();