Repository: apache/kafka
File: coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImplTest.java (Java)

Code before:
            ArgumentCaptor<MetricName> metricNameCaptor = ArgumentCaptor.forClass(MetricName.class);
            verify(metrics, atLeastOnce()).addMetric(metricNameCaptor.capture(), any(MetricValueProvider.class));
            metricNames = new HashSet<>(metricNameCaptor.getAllValues());

            // Check that all gauges were registered.
            expectedMetricNames(metrics).forEach(metricName -> assertTrue(metrics.metrics().containsKey(metricName)));
        }

        clearInvocations(metrics);

        // Create second CoordinatorRuntimeMetricsImpl instance and capture sensor and metric names.
        Set<String> otherSensorNames;
        Set<MetricName> otherMetricNames;
        try (CoordinatorRuntimeMetricsImpl otherRuntimeMetrics = new CoordinatorRuntimeMetricsImpl(metrics, OTHER_METRICS_GROUP)) {
            otherRuntimeMetrics.registerEventQueueSizeGauge(() -> 0);

            ArgumentCaptor<String> sensorCaptor = ArgumentCaptor.forClass(String.class);
            verify(metrics, atLeastOnce()).sensor(sensorCaptor.capture());
            otherSensorNames = new HashSet<>(sensorCaptor.getAllValues());

            ArgumentCaptor<MetricName> metricNameCaptor = ArgumentCaptor.forClass(MetricName.class);
            verify(metrics, atLeastOnce()).addMetric(metricNameCaptor.capture(), any(MetricValueProvider.class));
            otherMetricNames = new HashSet<>(metricNameCaptor.getAllValues());

            // Check that all gauges were registered.
            assertEquals(sensorNames.size(), otherSensorNames.size());
            assertEquals(metricNames.size(), otherMetricNames.size());
        }

        // Check for shared sensors.
        Set<String> sharedSensorNames = new HashSet<>(sensorNames);
        sharedSensorNames.retainAll(otherSensorNames);
        assertTrue(sharedSensorNames.isEmpty(),
            "Found shared sensors between two CoordinatorRuntimeMetricsImpl instances: " + sharedSensorNames);

        // Check for shared metrics.
        Set<MetricName> sharedMetricNames = new HashSet<>(metricNames);
        sharedMetricNames.retainAll(otherMetricNames);
        assertTrue(sharedMetricNames.isEmpty(),
            "Found shared metrics between two CoordinatorRuntimeMetricsImpl instances: " + sharedMetricNames);
    }

    @Test
    public void testUpdateNumPartitionsMetrics() {
        Metrics metrics = new Metrics();

        try (CoordinatorRuntimeMetricsImpl runtimeMetrics = new CoordinatorRuntimeMetricsImpl(metrics, METRICS_GROUP)) {
            IntStream.range(0, 10)
                .forEach(__ -> runtimeMetrics.recordPartitionStateChange(CoordinatorState.INITIAL, CoordinatorState.LOADING));
            IntStream.range(0, 8)
                .forEach(__ -> runtimeMetrics.recordPartitionStateChange(CoordinatorState.LOADING, CoordinatorState.ACTIVE));

Review comment (suggestion):
We can't use `expectedMetricNames` here because it's tied to the `METRICS_GROUP` metrics group.

Code after:
            verify(metrics, atLeastOnce()).addMetric(metricNameCaptor.capture(), any(MetricValueProvider.class));
            metricNames = new HashSet<>(metricNameCaptor.getAllValues());

            // Check that all gauges were registered.
            expectedMetricNames(metrics).forEach(metricName -> assertTrue(metrics.metrics().containsKey(metricName)));
        }

        clearInvocations(metrics);

        // Create second CoordinatorRuntimeMetricsImpl instance and capture sensor and metric names.
        Set<String> otherSensorNames;
        Set<MetricName> otherMetricNames;
        try (CoordinatorRuntimeMetricsImpl otherRuntimeMetrics = new CoordinatorRuntimeMetricsImpl(metrics, OTHER_METRICS_GROUP)) {
            otherRuntimeMetrics.registerEventQueueSizeGauge(() -> 0);

            ArgumentCaptor<String> sensorCaptor = ArgumentCaptor.forClass(String.class);
            verify(metrics, atLeastOnce()).sensor(sensorCaptor.capture());
            otherSensorNames = new HashSet<>(sensorCaptor.getAllValues());

            ArgumentCaptor<MetricName> metricNameCaptor = ArgumentCaptor.forClass(MetricName.class);
            verify(metrics, atLeastOnce()).addMetric(metricNameCaptor.capture(), any(MetricValueProvider.class));
            otherMetricNames = new HashSet<>(metricNameCaptor.getAllValues());

            // Check that all gauges were registered.
            assertEquals(sensorNames.size(), otherSensorNames.size());
            assertEquals(metricNames.size(), otherMetricNames.size());
        }

        // Check for shared sensors.
        Set<String> sharedSensorNames = new HashSet<>(sensorNames);
        sharedSensorNames.retainAll(otherSensorNames);
        assertTrue(sharedSensorNames.isEmpty(),
            "Found shared sensors between two CoordinatorRuntimeMetricsImpl instances: " + sharedSensorNames);

        // Check for shared metrics.
        Set<MetricName> sharedMetricNames = new HashSet<>(metricNames);
        sharedMetricNames.retainAll(otherMetricNames);
        assertTrue(sharedMetricNames.isEmpty(),
            "Found shared metrics between two CoordinatorRuntimeMetricsImpl instances: " + sharedMetricNames);
    }

    @Test
    public void testUpdateNumPartitionsMetrics() {
        Metrics metrics = new Metrics();

        try (CoordinatorRuntimeMetricsImpl runtimeMetrics = new CoordinatorRuntimeMetricsImpl(metrics, METRICS_GROUP)) {
            IntStream.range(0, 10)
                .forEach(__ -> runtimeMetrics.recordPartitionStateChange(CoordinatorState.INITIAL, CoordinatorState.LOADING));
            IntStream.range(0, 8)
                .forEach(__ -> runtimeMetrics.recordPartitionStateChange(CoordinatorState.LOADING, CoordinatorState.ACTIVE));
            IntStream.range(0, 8)