content: /*
   * 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.streams;
  
  import org.apache.kafka.clients.admin.AdminClientConfig;
  import org.apache.kafka.clients.admin.GroupListing;
  import org.apache.kafka.clients.consumer.ConsumerConfig;
  import org.apache.kafka.common.GroupState;
  import org.apache.kafka.common.GroupType;
  import org.apache.kafka.common.serialization.Serdes;
  import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
  import org.apache.kafka.streams.GroupProtocol;
  import org.apache.kafka.streams.KafkaStreams;
  import org.apache.kafka.streams.StreamsBuilder;
  import org.apache.kafka.streams.StreamsConfig;
  import org.apache.kafka.streams.Topology;
  import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
  import org.apache.kafka.streams.kstream.Consumed;
  import org.apache.kafka.streams.kstream.Produced;
  import org.apache.kafka.test.TestUtils;
  import org.apache.kafka.tools.ToolsTestUtils;
  
  import org.junit.jupiter.api.AfterAll;
  import org.junit.jupiter.api.Assertions;
  import org.junit.jupiter.api.BeforeAll;
  import org.junit.jupiter.api.Tag;
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.api.Timeout;
  
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Locale;
  import java.util.Objects;
  import java.util.Optional;
  import java.util.Properties;
  import java.util.Set;
  import java.util.concurrent.atomic.AtomicReference;
  import java.util.stream.Collectors;
  
  import joptsimple.OptionException;
  
  import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.startApplicationAndWaitUntilRunning;
  
  @Timeout(600)
  @Tag("integration")
  public class ListStreamsGroupTest {
  
      public static EmbeddedKafkaCluster cluster = null;
      static KafkaStreams streams;
      private static final String APP_ID = "streams-group-command-test";
      private static final String INPUT_TOPIC = "customInputTopic";
      private static final String OUTPUT_TOPIC = "customOutputTopic";
  
      @BeforeAll
      public static void setup() throws Exception {
          // start the cluster and create the input topic
          final Properties props = new Properties();
          props.setProperty(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, "classic,consumer,streams");
          cluster = new EmbeddedKafkaCluster(1, props);
          cluster.start();
          cluster.createTopic(INPUT_TOPIC, 2, 1);
  
  
          // start kafka streams
          Properties streamsProp = new Properties();
          streamsProp.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
          streamsProp.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers());
          streamsProp.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
          streamsProp.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
          streamsProp.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath());
          streamsProp.put(StreamsConfig.APPLICATION_ID_CONFIG, APP_ID);
          streamsProp.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 2);
          streamsProp.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name().toLowerCase(Locale.getDefault()));
  
          streams = new KafkaStreams(topology(), streamsProp);
          startApplicationAndWaitUntilRunning(streams);
      }
  
      @AfterAll
      public static void closeCluster() {
          streams.close();
          cluster.stop();
          cluster = null;
      }
  
      @Test
      public void testListStreamsGroupWithoutFilters() throws Exception {
          try (StreamsGroupCommand.StreamsGroupService service = getStreamsGroupService(new String[]{"--bootstrap-server", cluster.bootstrapServers(), "--list"})) {
              Set<String> expectedGroups = new HashSet<>(Collections.singleton(APP_ID));
  
              final AtomicReference<Set> foundGroups = new AtomicReference<>();
              TestUtils.waitForCondition(() -> {
                  foundGroups.set(new HashSet<>(service.listStreamsGroups()));
                  return Objects.equals(expectedGroups, foundGroups.get());
              }, "Expected --list to show streams groups " + expectedGroups + ", but found " + foundGroups.get() + ".");
  
          }
      }
  
      @Test
      public void testListWithUnrecognizedNewOption() {
          String[] cgcArgs = new String[]{"--new-option", "--bootstrap-server", cluster.bootstrapServers(), "--list"};
          Assertions.assertThrows(OptionException.class, () -> getStreamsGroupService(cgcArgs));
      }
  
      @Test
      public void testListStreamsGroupWithStates() throws Exception {
          try (StreamsGroupCommand.StreamsGroupService service = getStreamsGroupService(new String[]{"--bootstrap-server", cluster.bootstrapServers(), "--list", "--state"})) {
              Set<GroupListing> expectedListing = Set.of(
                  new GroupListing(
                      APP_ID,
                      Optional.of(GroupType.STREAMS),
                      "streams",
                      Optional.of(GroupState.STABLE))
              );
  
              final AtomicReference<Set<GroupListing>> foundListing = new AtomicReference<>();
  
              TestUtils.waitForCondition(() -> {
                  foundListing.set(new HashSet<>(service.listStreamsGroupsInStates(Collections.emptySet())));
                  return Objects.equals(expectedListing, foundListing.get());
              }, "Expected --list to show streams groups " + expectedListing + ", but found " + foundListing.get() + ".");
          }
      }
  
      @Test
      public void testListStreamsGroupWithSpecifiedStates() throws Exception {
          try (StreamsGroupCommand.StreamsGroupService service = getStreamsGroupService(new String[]{"--bootstrap-server", cluster.bootstrapServers(), "--list", "--state", "stable"})) {
              Set<GroupListing> expectedListing = Set.of(
                  new GroupListing(
                      APP_ID,
                      Optional.of(GroupType.STREAMS),
                      "streams",
                      Optional.of(GroupState.STABLE))
              );
  
              final AtomicReference<Set<GroupListing>> foundListing = new AtomicReference<>();
  
              TestUtils.waitForCondition(() -> {
                  foundListing.set(new HashSet<>(service.listStreamsGroupsInStates(Collections.emptySet())));
                  return Objects.equals(expectedListing, foundListing.get());
              }, "Expected --list to show streams groups " + expectedListing + ", but found " + foundListing.get() + ".");
          }
  
          try (StreamsGroupCommand.StreamsGroupService service = getStreamsGroupService(new String[]{"--bootstrap-server", cluster.bootstrapServers(), "--list", "--state", "PreparingRebalance"})) {
              Set<GroupListing> expectedListing = Collections.emptySet();
  
              final AtomicReference<Set<GroupListing>> foundListing = new AtomicReference<>();
  
              TestUtils.waitForCondition(() -> {
                  foundListing.set(new HashSet<>(service.listStreamsGroupsInStates(Collections.singleton(GroupState.PREPARING_REBALANCE))));
                  return Objects.equals(expectedListing, foundListing.get());
              }, "Expected --list to show streams groups " + expectedListing + ", but found " + foundListing.get() + ".");
          }
      }
  
      @Test
      public void testListStreamsGroupOutput() throws Exception {
          validateListOutput(
              Arrays.asList("--bootstrap-server", cluster.bootstrapServers(), "--list"),
              Collections.emptyList(),
              Set.of(Collections.singletonList(APP_ID))
          );
  
          validateListOutput(
              Arrays.asList("--bootstrap-server", cluster.bootstrapServers(), "--list", "--state"),
              Arrays.asList("GROUP", "STATE"),
              Set.of(Arrays.asList(APP_ID, "Stable"))
          );
  
          validateListOutput(
              Arrays.asList("--bootstrap-server", cluster.bootstrapServers(), "--list", "--state", "Stable"),
              Arrays.asList("GROUP", "STATE"),
              Set.of(Arrays.asList(APP_ID, "Stable"))
          );
  
          // Check case-insensitivity in state filter.
          validateListOutput(
              Arrays.asList("--bootstrap-server", cluster.bootstrapServers(), "--list", "--state", "stable"),
              Arrays.asList("GROUP", "STATE"),
              Set.of(Arrays.asList(APP_ID, "Stable"))
          );
      }
  
      private static Topology topology() {
          final StreamsBuilder builder = new StreamsBuilder();
          builder.stream(INPUT_TOPIC, Consumed.with(Serdes.String(), Serdes.String()))
              .flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.getDefault()).split("\\W+")))
              .groupBy((key, value) -> value)
              .count()
              .toStream().to(OUTPUT_TOPIC, Produced.with(Serdes.String(), Serdes.Long()));
          return builder.build();
      }
  
      private StreamsGroupCommand.StreamsGroupService getStreamsGroupService(String[] args) {
          StreamsGroupCommandOptions opts = StreamsGroupCommandOptions.fromArgs(args);
          return new StreamsGroupCommand.StreamsGroupService(
              opts,
              Collections.singletonMap(AdminClientConfig.RETRIES_CONFIG, Integer.toString(Integer.MAX_VALUE))
          );
      }
  
      private static void validateListOutput(
          List<String> args,
          List<String> expectedHeader,
          Set<List<String>> expectedRows
      ) throws InterruptedException {
          final AtomicReference<String> out = new AtomicReference<>("");
          TestUtils.waitForCondition(() -> {
              String output = ToolsTestUtils.grabConsoleOutput(() -> StreamsGroupCommand.main(args.toArray(new String[0])));
              out.set(output);
  
              String[] lines = output.split("\n");
              if (lines.length == 1 && lines[0].isEmpty()) lines = new String[]{};
  
              if (!expectedHeader.isEmpty() && lines.length > 0) {
                  List<String> header = Arrays.asList(lines[0].split("\\s+"));
                  if (!expectedHeader.equals(header)) return false;
              }
  
              Set<List<String>> groups = Arrays.stream(lines, expectedHeader.isEmpty() ? 0 : 1, lines.length)
                  .map(line -> Arrays.asList(line.split("\\s+")))
                  .collect(Collectors.toSet());
              return expectedRows.equals(groups);
          }, () -> String.format("Expected header=%s and groups=%s, but found:%n%s", expectedHeader, expectedRows, out.get()));
      }
  }