Repository: apache/kafka
File: tools/src/test/java/org/apache/kafka/tools/streams/DeleteStreamsGroupTest.java (Java)

Code before:
            "The internal topics of the streams group " + appId2 + " were not deleted as expected.");
        TestUtils.waitForCondition(() -> getInternalTopics(appId3).isEmpty(),
            "The internal topics of the streams group " + appId3 + " were not deleted as expected.");
    }

    @Test
    public void testDeleteAllGroupsAfterVersionDowngrade() throws Exception {
        final String appId = generateGroupAppId();
        String[] args = new String[]{"--bootstrap-server", bootstrapServers, "--delete", "--all-groups", "--delete-all-internal-topics"};

        StreamsGroupCommand.StreamsGroupService service = getStreamsGroupService(args);
        try (KafkaStreams streams = startKSApp(appId, service)) {
            stopKSApp(appId, streams, service);
            // downgrade the streams.version to 0
            updateStreamsGroupProtocol((short) 0);
            final Map<String, Throwable> result = new HashMap<>();
            String output = ToolsTestUtils.grabConsoleOutput(() -> result.putAll(service.deleteGroups()));
            System.out.println(output);

            assertTrue(output.contains("Deletion of requested streams groups ('" + appId + "') was successful."),
                "The streams group could not be deleted as expected");
            assertTrue(output.contains("Retrieving internal topics is not supported by the broker version. "));
            assertTrue(output.contains("Use 'kafka-topics.sh' to delete the group's internal topics."));
            // Validate the list of internal topics in error message
            assertTrue(output.contains("Internal topics:"));
            assertTrue(output.contains("aggregated_value-changelog"));
            assertTrue(output.contains("aggregated_value-repartition"));
            assertTrue(output.contains("count-repartition"));

            assertEquals(1, result.size());
            assertTrue(result.containsKey(appId));
            assertNull(result.get(appId), "The streams group could not be deleted as expected");
            assertEquals(3, getInternalTopics(appId).size(),
                "The internal topics were deleted, but they shouldn't have been.");
        } finally {
            // upgrade back the streams.version to 1
            updateStreamsGroupProtocol((short) 1);
        }
    }

    private Set<String> getInternalTopics(String appId) {
        try {
            Set<String> topics = adminClient.listTopics().names().get();
            return topics.stream()
                .filter(topic -> topic.startsWith(appId + "-"))
                .filter(topic -> topic.endsWith("-changelog") || topic.endsWith("-repartition"))
                .collect(Collectors.toSet());
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }
    }

Review comment (suggestion):
@lucliu1108 Perfect
Btw, could we add the appId/groupId add the beginning? Sth that shows the internal topic names name pattern? thanks.

Code after:
            "The internal topics of the streams group " + appId2 + " were not deleted as expected.");
        TestUtils.waitForCondition(() -> getInternalTopics(appId3).isEmpty(),
            "The internal topics of the streams group " + appId3 + " were not deleted as expected.");
    }

    @Test
    public void testDeleteAllGroupsAfterVersionDowngrade() throws Exception {
        final String appId = generateGroupAppId();
        String[] args = new String[]{"--bootstrap-server", bootstrapServers, "--delete", "--all-groups", "--delete-all-internal-topics"};

        StreamsGroupCommand.StreamsGroupService service = getStreamsGroupService(args);
        try (KafkaStreams streams = startKSApp(appId, service)) {
            stopKSApp(appId, streams, service);
            // downgrade the streams.version to 0
            updateStreamsGroupProtocol((short) 0);
            final Map<String, Throwable> result = new HashMap<>();
            String output = ToolsTestUtils.grabConsoleOutput(() -> result.putAll(service.deleteGroups()));
            System.out.println(output);

            assertTrue(output.contains("Deletion of requested streams groups ('" + appId + "') was successful."),
                "The streams group could not be deleted as expected");
            assertTrue(output.contains("Retrieving internal topics is not supported by the broker version."));
            assertTrue(output.contains("Use 'kafka-topics.sh' to delete the group's internal topics."));
            // Validate the list of internal topics in error message
            assertTrue(output.contains("Internal topics:"));
            System.out.println(output);
            assertTrue(
                output.matches("(?s).*" + APP_ID_PREFIX + "[a-zA-Z0-9\\-]+-(aggregated_value-changelog|repartition|changelog).*"),
                "The internal topic name does not match the expected format. Output: " + output
            );

            assertEquals(1, result.size());
            assertTrue(result.containsKey(appId));
            assertNull(result.get(appId), "The streams group could not be deleted as expected");
            assertEquals(3, getInternalTopics(appId).size(),
                "The internal topics were deleted, but they shouldn't have been.");
        } finally {
            // upgrade back the streams.version to 1
            updateStreamsGroupProtocol((short) 1);
        }
    }

    private Set<String> getInternalTopics(String appId) {
        try {
            Set<String> topics = adminClient.listTopics().names().get();
            return topics.stream()
                .filter(topic -> topic.startsWith(appId + "-"))
                .filter(topic -> topic.endsWith("-changelog") || topic.endsWith("-repartition"))
                .collect(Collectors.toSet());
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);