Repository: apache/kafka
File: streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KafkaStreamsTelemetryIntegrationTest.java (Java)

Code before:
            final Map<MetricName, ? extends Metric> embeddedConsumerMetrics = INTERCEPTING_CONSUMERS.get(FIRST_INSTANCE_CLIENT).metrics();
            final Map<MetricName, ? extends Metric> embeddedAdminMetrics = INTERCEPTING_ADMIN_CLIENTS.get(FIRST_INSTANCE_CLIENT).metrics();

            streamsThreadMetrics.forEach(metricName -> assertFalse(embeddedConsumerMetrics.containsKey(metricName), "Stream thread metric found in client metrics" + metricName));
            streamsClientMetrics.forEach(metricName -> assertFalse(embeddedAdminMetrics.containsKey(metricName), "Stream client metric found in client metrics" + metricName));
        }
    }

    private void subscribeForStreamsMetrics() throws Exception {
        final Properties clientProps = new Properties();
        clientProps.put("bootstrap.servers", cluster.bootstrapServers());
        try (final ClientMetricsCommand.ClientMetricsService clientMetricsService = new ClientMetricsCommand.ClientMetricsService(clientProps)) {
            final String[] metricsSubscriptionParameters = new String[]{"--bootstrap-server", cluster.bootstrapServers(), "--metrics", "org.apache.kafka.stream", "--alter", "--name", "streams-task-metrics-subscription", "--interval", "1000"};
            final ClientMetricsCommand.ClientMetricsCommandOptions commandOptions = new ClientMetricsCommand.ClientMetricsCommandOptions(metricsSubscriptionParameters);
            clientMetricsService.alterClientMetrics(commandOptions);
        }
    }

    private List<String> getTaskIdsAsStrings(final KafkaStreams streams) {
        return streams.metadataForLocalThreads().stream()
                .flatMap(threadMeta -> threadMeta.activeTasks().stream()
                        .map(taskMeta -> taskMeta.taskId().toString()))
                .toList();
    }

    private static Stream<Arguments> singleAndMultiTaskParameters() {
        return Stream.of(
            Arguments.of("simple", true, "classic"),
            Arguments.of("simple", false, "classic"),
            Arguments.of("complex", true, "classic"),
            Arguments.of("complex", false, "classic"),
            Arguments.of("simple", true, "streams"),
            Arguments.of("simple", false, "streams")
        );
    }

    private static Stream<Arguments> multiTaskParameters() {
        return Stream.of(
            Arguments.of(true, "classic"),
            Arguments.of(false, "classic")
        );
    }

    private Properties props(final boolean stateUpdaterEnabled, final String groupProtocol) {
        return props(mkObjectProperties(mkMap(
            mkEntry(StreamsConfig.InternalConfig.STATE_UPDATER_ENABLED, stateUpdaterEnabled),
            mkEntry(StreamsConfig.GROUP_PROTOCOL_CONFIG, groupProtocol)
        )));
    }

    private Properties props(final Properties extraProperties) {

Review comment (performance):
Renamed to `topologyComplexityAndRebalanceProtocol` as per https://github.com/apache/kafka/pull/19275#discussion_r2107363103

Code after:
            final Map<MetricName, ? extends Metric> embeddedConsumerMetrics = INTERCEPTING_CONSUMERS.get(FIRST_INSTANCE_CLIENT).metrics();
            final Map<MetricName, ? extends Metric> embeddedAdminMetrics = INTERCEPTING_ADMIN_CLIENTS.get(FIRST_INSTANCE_CLIENT).metrics();

            streamsThreadMetrics.forEach(metricName -> assertFalse(embeddedConsumerMetrics.containsKey(metricName), "Stream thread metric found in client metrics" + metricName));
            streamsClientMetrics.forEach(metricName -> assertFalse(embeddedAdminMetrics.containsKey(metricName), "Stream client metric found in client metrics" + metricName));
        }
    }

    private void subscribeForStreamsMetrics() throws Exception {
        final Properties clientProps = new Properties();
        clientProps.put("bootstrap.servers", cluster.bootstrapServers());
        try (final ClientMetricsCommand.ClientMetricsService clientMetricsService = new ClientMetricsCommand.ClientMetricsService(clientProps)) {
            final String[] metricsSubscriptionParameters = new String[]{"--bootstrap-server", cluster.bootstrapServers(), "--metrics", "org.apache.kafka.stream", "--alter", "--name", "streams-task-metrics-subscription", "--interval", "1000"};
            final ClientMetricsCommand.ClientMetricsCommandOptions commandOptions = new ClientMetricsCommand.ClientMetricsCommandOptions(metricsSubscriptionParameters);
            clientMetricsService.alterClientMetrics(commandOptions);
        }
    }

    private List<String> getTaskIdsAsStrings(final KafkaStreams streams) {
        return streams.metadataForLocalThreads().stream()
                .flatMap(threadMeta -> threadMeta.activeTasks().stream()
                        .map(taskMeta -> taskMeta.taskId().toString()))
                .toList();
    }

    private static Stream<Arguments> topologyComplexityAndRebalanceProtocol() {
        return Stream.of(
            Arguments.of("simple", "classic"),
            Arguments.of("complex", "classic"),
            Arguments.of("simple", "streams")
        );
    }

    private Properties props(final String groupProtocol) {
        return props(mkObjectProperties(mkMap(
            mkEntry(StreamsConfig.GROUP_PROTOCOL_CONFIG, groupProtocol)
        )));
    }

    private Properties props(final Properties extraProperties) {
        final Properties streamsConfiguration = new Properties();
        streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, appId);
        streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers());
        streamsConfiguration.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0);
        streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(appId).getPath());
        streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
        streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
        streamsConfiguration.put(StreamsConfig.DEFAULT_CLIENT_SUPPLIER_CONFIG, TestClientSupplier.class);
        streamsConfiguration.put(StreamsConfig.InternalConfig.INTERNAL_CONSUMER_WRAPPER, TestConsumerWrapper.class);
        streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        streamsConfiguration.putAll(extraProperties);