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

Code before:
            final List<String> actualInstanceMetrics = TelemetryPlugin.SUBSCRIBED_METRICS.get(adminInstanceId);
            final List<String> expectedInstanceMetrics = Arrays.asList(
                "org.apache.kafka.stream.alive.stream.threads",
                "org.apache.kafka.stream.client.state",
                "org.apache.kafka.stream.failed.stream.threads",
                "org.apache.kafka.stream.recording.level");
            
            assertEquals(expectedInstanceMetrics, actualInstanceMetrics);

            TestUtils.waitForCondition(() -> TelemetryPlugin.processId != null,
                    30_000,
                    "Never received the process id");

            assertEquals(expectedProcessId, TelemetryPlugin.processId);
        }
    }

    @ParameterizedTest
    @MethodSource("singleAndMultiTaskParameters")
    public void shouldPassMetrics(final String topologyType, final boolean stateUpdaterEnabled, final String groupProtocol) throws Exception {
        // Streams metrics should get passed to Admin and Consumer
        streamsApplicationProperties = props(stateUpdaterEnabled, groupProtocol);
        final Topology topology = topologyType.equals("simple") ? simpleTopology(false) : complexTopology();

        final AtomicInteger runningStateCount = new AtomicInteger(0);
        final int expectedRunningStateCount = DEFAULT_GROUP_PROTOCOL.equals(groupProtocol) || "simple".equals(topologyType) ? 1 : 2;

        try (final KafkaStreams streams = new KafkaStreams(topology, streamsApplicationProperties)) {
            streams.setStateListener((newState, oldState) -> {
                if (newState == KafkaStreams.State.RUNNING) {
                    runningStateCount.incrementAndGet();
                }
            });
            streams.start();
            waitForCondition(
                () -> runningStateCount.get() == expectedRunningStateCount,
                "Application did not get into RUNNING state as expected."
            );

            final List<MetricName> streamsThreadMetrics = streams.metrics().values().stream().map(Metric::metricName)
                    .filter(metricName -> metricName.tags().containsKey("thread-id")).toList();

            final List<MetricName> streamsClientMetrics = streams.metrics().values().stream().map(Metric::metricName)
                    .filter(metricName -> metricName.group().equals("stream-metrics")).toList();

            final List<MetricName> consumerPassedStreamThreadMetricNames = INTERCEPTING_CONSUMERS.get(FIRST_INSTANCE_CLIENT).passedMetrics().stream().map(KafkaMetric::metricName).toList();
            final List<MetricName> adminPassedStreamClientMetricNames = INTERCEPTING_ADMIN_CLIENTS.get(FIRST_INSTANCE_CLIENT).passedMetrics.stream().map(KafkaMetric::metricName).toList();


            assertEquals(streamsThreadMetrics.size(), consumerPassedStreamThreadMetricNames.size());
            consumerPassedStreamThreadMetricNames.forEach(metricName -> assertTrue(streamsThreadMetrics.contains(metricName), "Streams metrics doesn't contain " + metricName));

Review comment (bug):
It's not clear to my, why "simple" topologies show different behavior, compare to "complex" ones.

When I run the code for "simple" I get the follow log from the consumer:
```[2025-07-10 14:28:52,035] INFO [Consumer clientId=shouldPassMetricsBi3tKRB2SZOSy_YU2ch9rw-5921224d-0525-4866-9db4-71f9f95c99cd-StreamThread-1-consumer, groupId=shouldPassMetricsBi3tKRB2SZOSy_YU2ch9rw] Assigned tasks with local epoch 0
	Member:                        dL_1PoEdQ4C9BhYUVUzFUQ
	Assigned active tasks:         [0_0]
	Owned active tasks:            []
	Active tasks to revoke:        []
	Assigned standby tasks:        []
	Owned standby tasks:           []
	Assigned warm-up tasks:        []
	Owned warm-up tasks:           []
 (org.apache.kafka.clients.consumer.internals.StreamsMembershipManager:1029)
 ```
 
 As you can see, the task is assigned and we only get a single rebalance.
 
 However, when the test executed for "complex" topology, the consumer first logs:
 ```
 [2025-07-10 14:29:23,384] INFO [Consumer clientId=shouldPassMetricsx1__GHUPQvS0AkXE48xleQ-3f598d1a-1f8a-4923-9332-0d947bb38ab2-StreamThread-1-consumer, groupId=shouldPassMetricsx1__GHUPQvS0AkXE48xleQ] Assigned tasks with local epoch 0
	Member:                        mcJBdaR5TF2QjDoMvLXKBw
	Assigned active tasks:         []
	Owned active tasks:            []
	Active tasks to revoke:        []
	Assigned standby tasks:        []
	Owned standby tasks:           []
	Assigned warm-up tasks:        []
	Owned warm-up tasks:           []
 (org.apache.kafka.clients.consumer.internals.StreamsMembershipManager:1029)
 ```
 and there is no task assignment yet. Only a second rebalance will do the assignment:
 ```
 [2025-07-10 14:29:28,393] INFO [Consumer clientId=shouldPassMetricsx1__GHUPQvS0AkXE48xleQ-3f598d1a-1f8a-4923-9332-0d947bb38ab2-StreamThread-1-consumer, groupId=shouldPassMetricsx1__GHUPQvS0AkXE48xleQ] Assigned tasks with local epoch 1
	Member:                        mcJBdaR5TF2QjDoMvLXKBw
	Assigned active tasks:         [0_0, 0_1, 1_0, 1_1]
	Owned active tasks:            []
	Active tasks to revoke:        []
	Assigned standby tasks:        []
	Owned standby tasks:           []
	Assigned warm-up tasks:        []
	Owned warm-up tasks:           []
 (org.apache.kafka.clients.consumer.internals.StreamsMembershipManager:1029)
 ```

Why do we get the assignment for the simple case in the first rebalance already, but not for the second case? Is it because the complex topology needs to create a repartition topic, and thus, does not even try to compute an assignment on the first HB?

I am worried that there might be some race condition that is not covered by this test fix.

Code after:
                "org.apache.kafka.stream.alive.stream.threads",
                "org.apache.kafka.stream.client.state",
                "org.apache.kafka.stream.failed.stream.threads",
                "org.apache.kafka.stream.recording.level");
            
            assertEquals(expectedInstanceMetrics, actualInstanceMetrics);

            TestUtils.waitForCondition(() -> TelemetryPlugin.processId != null,
                    30_000,
                    "Never received the process id");

            assertEquals(expectedProcessId, TelemetryPlugin.processId);
        }
    }

    @ParameterizedTest
    @MethodSource("singleAndMultiTaskParameters")
    public void shouldPassMetrics(final String topologyType, final boolean stateUpdaterEnabled, final String groupProtocol) throws Exception {
        // Streams metrics should get passed to Admin and Consumer
        streamsApplicationProperties = props(stateUpdaterEnabled, groupProtocol);
        final Topology topology = topologyType.equals("simple") ? simpleTopology(false) : complexTopology();

        try (final KafkaStreams streams = new KafkaStreams(topology, streamsApplicationProperties)) {
            IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);

            final List<MetricName> streamsThreadMetrics = streams.metrics().values().stream().map(Metric::metricName)
                    .filter(metricName -> metricName.tags().containsKey("thread-id")).toList();

            final List<MetricName> streamsClientMetrics = streams.metrics().values().stream().map(Metric::metricName)
                    .filter(metricName -> metricName.group().equals("stream-metrics")).toList();



            final List<MetricName> consumerPassedStreamThreadMetricNames = INTERCEPTING_CONSUMERS.get(FIRST_INSTANCE_CLIENT).passedMetrics().stream().map(KafkaMetric::metricName).toList();
            final List<MetricName> adminPassedStreamClientMetricNames = INTERCEPTING_ADMIN_CLIENTS.get(FIRST_INSTANCE_CLIENT).passedMetrics.stream().map(KafkaMetric::metricName).toList();


            assertEquals(streamsThreadMetrics.size(), consumerPassedStreamThreadMetricNames.size());
            consumerPassedStreamThreadMetricNames.forEach(metricName -> assertTrue(streamsThreadMetrics.contains(metricName), "Streams metrics doesn't contain " + metricName));

            assertEquals(streamsClientMetrics.size(), adminPassedStreamClientMetricNames.size());
            adminPassedStreamClientMetricNames.forEach(metricName -> assertTrue(streamsClientMetrics.contains(metricName), "Client metrics doesn't contain " + metricName));
        }
    }

    @ParameterizedTest
    @MethodSource("multiTaskParameters")
    public void shouldPassCorrectMetricsDynamicInstances(final boolean stateUpdaterEnabled, final String groupProtocol) throws Exception {
        // Correct streams metrics should get passed with dynamic membership
        streamsApplicationProperties = props(stateUpdaterEnabled, groupProtocol);
        streamsApplicationProperties.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(appId).getPath() + "-ks1");