aliehsaeedii commented on code in PR #18911: URL: https://github.com/apache/kafka/pull/18911#discussion_r1959494561
########## tools/src/test/java/org/apache/kafka/tools/streams/StreamsGroupCommandTest.java: ########## @@ -0,0 +1,447 @@ +/* + * 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 StreamsGroupCommandTest { + + 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.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() throws Exception { + 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")) + ); + } + + @Test + public void testDescribeStreamsGroup() throws Exception { + final List<String> expectedHeaders = List.of("GROUP", "TOPIC", "PARTITION", "OFFSET-LAG"); + final Set<List<String>> expectedRows = Set.of(List.of(APP_ID, "", "0", "0"), List.of(APP_ID, "", "1", "0")); + final List<Integer> dontCares = List.of(1); + + + validateDescribeOutput( + Arrays.asList("--bootstrap-server", cluster.bootstrapServers(), "--describe"), expectedHeaders, expectedRows, 5, dontCares); + + // --describe --offsets has the same output as --describe + validateDescribeOutput( + Arrays.asList("--bootstrap-server", cluster.bootstrapServers(), "--describe", "--offsets"), expectedHeaders, expectedRows, 5, dontCares); + } + + @Test + public void testDescribeStreamsGroupWithVerboseOption() throws Exception { + final List<String> expectedHeaders = List.of("GROUP", "TOPIC", "PARTITION", "LEADER-EPOCH", "OFFSET-LAG"); + final Set<List<String>> expectedRows = Set.of(List.of(APP_ID, "", "0", "2", "0"), List.of(APP_ID, "", "1", "2", "0")); + final List<Integer> dontCares = List.of(1); + + validateDescribeOutput( + Arrays.asList("--bootstrap-server", cluster.bootstrapServers(), "--describe", "--verbose"), expectedHeaders, expectedRows, 5, dontCares); + // --describe --offsets has the same output as --describe + validateDescribeOutput( + Arrays.asList("--bootstrap-server", cluster.bootstrapServers(), "--describe", "--offsets", "--verbose"), expectedHeaders, expectedRows, 5, dontCares); + validateDescribeOutput( + Arrays.asList("--bootstrap-server", cluster.bootstrapServers(), "--describe", "--verbose", "--offsets"), expectedHeaders, expectedRows, 5, dontCares); + } + + @Test + public void testDescribeStreamsGroupWithStateOption() throws Exception { + final List<String> expectedHeaders = Arrays.asList("GROUP", "COORDINATOR", "(ID)", "STATE", "#MEMBERS"); + final Set<List<String>> expectedRows = Set.of(Arrays.asList(APP_ID, "", "", "Stable", "1")); + final List<Integer> dontCares = List.of(1, 2); Review Comment: It could be different in each run/env. The output is something like `localhost:50236 (0)` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
