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

Code before:
/*
 * 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.coordinator.common.runtime;

import org.apache.kafka.common.utils.ThreadUtils;

import com.yammer.metrics.core.Histogram;
import com.yammer.metrics.core.MetricName;
import com.yammer.metrics.core.MetricsRegistry;

import org.junit.jupiter.api.Test;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;

public class HdrHistogramTest {

    private static final double[] QUANTILES = new double[]{0.5, 0.75, 0.95, 0.98, 0.99, 0.999};

    private static final long MAX_VALUE = TimeUnit.MINUTES.toMillis(1L);
    private static final long REGULAR_VALUE = 100L;
    private static final int NUM_SIGNIFICANT_DIGITS = 3;

    @Test
    public void testHdrVsYammerUniform() {
        List<Long> samples = new ArrayList<>();
        for (long i = 0; i <= MAX_VALUE; i++) { // 60k iterations with the current constants
            samples.add(i);
        }
        testHdrHistogramVsYammerHistogram(samples, "uniform");
    }

    @Test
    public void testHdrVsYammerNormal() {
        List<Long> samples = new ArrayList<>();
        for (long i = 0; i <= MAX_VALUE; i++) { // 60k iterations with the current constants
            samples.add(ThreadLocalRandom.current().nextLong(MAX_VALUE));
        }
        testHdrHistogramVsYammerHistogram(samples, "normal");
    }

    @Test
    public void testHdrVsYammerBimodal() {
        List<Long> samples = new ArrayList<>();
        for (long i = 0; i <= MAX_VALUE; i++) { // 60k iterations with the current constants
            if (i % 500 == 0) {
                // generate a tail-like latency value, and do it once in every 500 iterations, so that the
                // p999 value is expected to reflect it
                samples.add(ThreadLocalRandom.current().nextLong(MAX_VALUE));
            } else {
                // generate a normal latency value
                samples.add(ThreadLocalRandom.current().nextLong(REGULAR_VALUE));
            }
        }
        testHdrHistogramVsYammerHistogram(samples, "bimodal");
    }

    private void testHdrHistogramVsYammerHistogram(List<Long> samples, String distributionLabel) {
        HdrHistogram hdrHistogram = new HdrHistogram(MAX_VALUE, NUM_SIGNIFICANT_DIGITS);
        Histogram yammerHistogram = new MetricsRegistry().newHistogram(new MetricName("", "", ""),
            true);

        Collections.sort(samples);
        double[] expectedQuantileValues = new double[QUANTILES.length];
        for (int i = 0; i < QUANTILES.length; i++) {
            int quantileIndex = (int) (Math.ceil(samples.size() * QUANTILES[i])) - 1;
            expectedQuantileValues[i] = samples.get(quantileIndex);
        }
        Collections.shuffle(samples);

        for (long sample : samples) {
            hdrHistogram.record(sample);
            yammerHistogram.update(sample);
        }

        System.out.printf("Testing HdrHistogram vs Yammer histogram for %s distribution%n",
            distributionLabel);
        long now = System.currentTimeMillis();
        int numYammerWins = 0;
        for (int i = 0; i < QUANTILES.length; i++) {
            double quantile = QUANTILES[i];
            double hdrHistogramValue = hdrHistogram.measurePercentile(now, quantile * 100);
            double yammerHistogramValue = yammerHistogram.getSnapshot().getValue(quantile);
            double expectedValue = expectedQuantileValues[i];
            System.out.printf(
                "Values for quantile %f: HdrHistogram: %f, Yammer histogram: %f, Expected: %f%n",
                quantile, hdrHistogramValue, yammerHistogramValue, expectedValue);
            if (Math.abs(expectedValue - hdrHistogramValue) > Math.abs(
                expectedValue - yammerHistogramValue)) {
                numYammerWins++;
            }
        }
        System.out.printf("HdrHistogram was more accurate: %d out of %d times%n",
            QUANTILES.length - numYammerWins, QUANTILES.length);
        assertTrue(numYammerWins <= QUANTILES.length / 2);
    }

    @Test
    public void testCount() throws Exception {
        int numUpdates = 100_000;
        HdrHistogram hdrHistogram = new HdrHistogram(
            MAX_VALUE, NUM_SIGNIFICANT_DIGITS);
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        for (int i = 0; i < numUpdates; i++) {
            executorService.submit(() -> hdrHistogram.record(1L));
        }
        executorService.shutdown();
        assertTrue(executorService.awaitTermination(1L, TimeUnit.SECONDS));
        assertEquals(numUpdates, hdrHistogram.count(System.currentTimeMillis()));
    }

    @Test
    public void testMax() throws Exception {
        int numSignificantDigits = 5;
        int numUpdates = 30_000;
        double expectedMax = 0.0;
        long[] values = new long[numUpdates];
        for (int i = 0; i < numUpdates; i++) {
            long value = ThreadLocalRandom.current().nextLong(MAX_VALUE);
            values[i] = value;
            expectedMax = Math.max(expectedMax, value);
        }

        HdrHistogram hdrHistogram = new HdrHistogram(MAX_VALUE, numSignificantDigits);
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        for (int i = 0; i < numUpdates; i++) {
            final long value = values[i];
            executorService.submit(() -> hdrHistogram.record(value));
        }
        executorService.shutdown();
        assertTrue(executorService.awaitTermination(1L, TimeUnit.SECONDS));

        double histogramMax = hdrHistogram.max(System.currentTimeMillis());
        assertEquals(expectedMax, histogramMax);
    }

    @Test
    public void testHistogramDataReset() {
        long maxSnapshotAgeMs = 10L;
        HdrHistogram hdrHistogram = new HdrHistogram(maxSnapshotAgeMs, MAX_VALUE, 3);
        int numEventsInFirstCycle = 1000;
        for (int i = 0; i < numEventsInFirstCycle; i++) {
            hdrHistogram.record(i);
        }
        long now = System.currentTimeMillis();
        assertEquals(numEventsInFirstCycle, hdrHistogram.count(now));
        int numEventsInSecondCycle = 2000;
        for (int i = 0; i < numEventsInSecondCycle; i++) {
            hdrHistogram.record(i);
        }
        assertEquals(numEventsInFirstCycle, hdrHistogram.count(now));
        assertEquals(numEventsInFirstCycle, hdrHistogram.count(now + maxSnapshotAgeMs));
        assertEquals(numEventsInSecondCycle, hdrHistogram.count(now + 1 + maxSnapshotAgeMs));
    }

    @Test
    public void testLatestHistogramRace() throws InterruptedException, ExecutionException {
        long maxSnapshotAgeMs = 10L;
        long now = System.currentTimeMillis();
        HdrHistogram hdrHistogram = new HdrHistogram(maxSnapshotAgeMs, MAX_VALUE, 1);
        ExecutorService countExecutor = Executors.newFixedThreadPool(2);
        for (int i = 1; i < 10000; i++) {
            int numEvents = 2;
            for (int j = 0; j < numEvents; j++) {
                hdrHistogram.record(i);
            }
            final long moreThanMaxAge = now + maxSnapshotAgeMs + 1;
            now = moreThanMaxAge;
            CountDownLatch latch = new CountDownLatch(1);
            Callable<Long> countTask = () -> {
                try {
                    assertTrue(latch.await(500, TimeUnit.MILLISECONDS));
                    return hdrHistogram.count(moreThanMaxAge);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            };
            Future<Long> t1Future = countExecutor.submit(countTask);
            Future<Long> t2Future = countExecutor.submit(countTask);
            latch.countDown();
            long t1Count = t1Future.get();
            long t2Count = t2Future.get();
            assertTrue(
                numEvents == t1Count && numEvents == t2Count,
                String.format("Expected %d events in both threads, got %d in T1 and %d in T2",
                    numEvents, t1Count, t2Count));
        }
        ThreadUtils.shutdownExecutorServiceQuietly(countExecutor, 500, TimeUnit.MILLISECONDS);
    }

    @Test
    public void testRecordLimit() {
        long highestTrackableValue = 10L;
        HdrHistogram hdrHistogram = new HdrHistogram(10L, highestTrackableValue, 3);

        hdrHistogram.record(highestTrackableValue + 1000L);
        assertEquals(highestTrackableValue, hdrHistogram.max(System.currentTimeMillis()));
    }
}


Review comment (suggestion):
There's no histogram `min` method. We can add one, but it'd be only used in this one test.

Code after:
                }
            };
            Future<Long> t1Future = countExecutor.submit(countTask);
            Future<Long> t2Future = countExecutor.submit(countTask);
            latch.countDown();
            long t1Count = t1Future.get();
            long t2Count = t2Future.get();
            assertTrue(
                numEvents == t1Count && numEvents == t2Count,
                String.format("Expected %d events in both threads, got %d in T1 and %d in T2",
                    numEvents, t1Count, t2Count));
        }
        ThreadUtils.shutdownExecutorServiceQuietly(countExecutor, 500, TimeUnit.MILLISECONDS);
    }

    @Test
    public void testRecordLimit() {
        long now = System.currentTimeMillis();
        long highestTrackableValue = 10L;
        HdrHistogram hdrHistogram = new HdrHistogram(10L, highestTrackableValue, 3);

        hdrHistogram.record(highestTrackableValue + 1000L);
        assertEquals(highestTrackableValue, hdrHistogram.max(now));

        hdrHistogram.record(-50L);
        assertEquals(0, hdrHistogram.max(now + 1000L));
    }
}