private Histogram aggregateHistograms(final Entry<String, List<Path>> entry) throws FileNotFoundException
{
    Histogram aggregate = null;

    for (final Path file : entry.getValue())
    {
        try (HistogramLogReader logReader = new HistogramLogReader(file.toFile()))
        {
            while (logReader.hasNext())
            {
                final Histogram histogram = (Histogram)logReader.nextIntervalHistogram();
                if (null == aggregate)
                {
                    aggregate = histogram;
                }
                else
                {
                    aggregate.add(histogram);
                }
            }
        }
    }

    return aggregate;
}
 

private Histogram readHistogram(final Path file) throws FileNotFoundException
{
    final List<EncodableHistogram> histograms = new ArrayList<>();
    final HistogramLogReader logReader = new HistogramLogReader(file.toFile());
    try
    {
        while (logReader.hasNext())
        {
            histograms.add(logReader.nextIntervalHistogram());
        }
    }
    finally
    {
        logReader.close();
    }
    assertEquals(1, histograms.size());
    return (Histogram)histograms.get(0);
}
 

@Test
public void testConvertHistogramToString() {
  Histogram histogram = new Histogram(1, 100, 3);
  histogram.recordValue(3);
  histogram.recordValue(25);
  histogram.recordValue(25);
  histogram.recordValue(92);
  String histogramString = KafkaExtractorStatsTracker.convertHistogramToString(histogram);

  HistogramLogReader logReader = new HistogramLogReader(new ByteArrayInputStream(histogramString.getBytes(
      Charsets.UTF_8)));
  Histogram histogram1 = (Histogram) logReader.nextIntervalHistogram();
  Assert.assertEquals(histogram1.getTotalCount(), 4);
  Assert.assertEquals(histogram1.getMaxValue(), 92);
  Assert.assertEquals(histogram1.getCountAtValue(25), 2);
  Assert.assertEquals(histogram1.getCountAtValue(3), 1);
  Assert.assertEquals(histogram1.getCountAtValue(92), 1);
}
 

private Histogram loadHistogram(final File file)
{
    try
    {
        return (Histogram) new HistogramLogReader(file).nextIntervalHistogram();
    }
    catch (FileNotFoundException e)
    {
        throw new RuntimeException("Could not process encoded histogram", e);
    }
}
 

private Histogram loadFromDisc(final String fileName) throws FileNotFoundException
{
    try (HistogramLogReader logReader = new HistogramLogReader(tempDir.resolve(fileName).toFile()))
    {
        return (Histogram)logReader.nextIntervalHistogram();
    }
}
 

public static void main(String[] args) throws FileNotFoundException {
    File inputFile = new File(args[0]);
    File outputFile = new File(inputFile.getParent(), inputFile.getName() + ".tmp");
    long startMillis = Long.parseLong(args[1]);
    long endMillis = Long.parseLong(args[2]);

    HistogramLogReader reader = new HistogramLogReader(inputFile);
    HistogramLogWriter writer = new HistogramLogWriter(outputFile);
    for (; ; ) {
        Histogram histogram = (Histogram) reader.nextIntervalHistogram();
        if (histogram == null) {
            break;
        }

        if (histogram.getStartTimeStamp() >= startMillis && histogram.getEndTimeStamp() <= endMillis) {
            Histogram out = new Histogram(
                    histogram.getLowestDiscernibleValue(),
                    histogram.getHighestTrackableValue(),
                    histogram.getNumberOfSignificantValueDigits());
            out.setStartTimeStamp(histogram.getStartTimeStamp());
            out.setEndTimeStamp(histogram.getEndTimeStamp());
            out.add(histogram);
            writer.outputIntervalHistogram(out);
        }
    }

    outputFile.renameTo(new File(args[0]));
}
 

public static Histogram getAccumulated(final File histogramFile) throws FileNotFoundException {
    Histogram accumulatedHistogram = null;
    DoubleHistogram accumulatedDoubleHistogram = null;

    HistogramLogReader histogramLogReader = new HistogramLogReader(histogramFile);

    int i = 0;
    while (histogramLogReader.hasNext()) {
        EncodableHistogram eh = histogramLogReader.nextIntervalHistogram();
        if (eh == null) {
            logger.error("The histogram library returned an unexpected null value");
            break;
        }

        if (i == 0) {
            if (eh instanceof DoubleHistogram) {
                accumulatedDoubleHistogram = ((DoubleHistogram) eh).copy();
                accumulatedDoubleHistogram.reset();
                accumulatedDoubleHistogram.setAutoResize(true);
            }
            else {
                accumulatedHistogram = ((Histogram) eh).copy();
                accumulatedHistogram.reset();
                accumulatedHistogram.setAutoResize(true);
            }
        }

        logger.debug("Processing histogram from point in time {} to {}",
                Instant.ofEpochMilli(eh.getStartTimeStamp()), Instant.ofEpochMilli(eh.getEndTimeStamp()));

        if (eh instanceof DoubleHistogram) {
            Objects.requireNonNull(accumulatedDoubleHistogram).add((DoubleHistogram) eh);
        }
        else {
            Objects.requireNonNull(accumulatedHistogram).add((Histogram) eh);
        }

        i++;
    }

    if (accumulatedHistogram == null) {
        throw new EmptyDataSet("The HDR data file did not contain any histogram data");
    }

    return accumulatedHistogram;
}
 

@Test(timeout = 120_000L)
public void shouldWriteLatencies() throws IOException, InterruptedException {
    final int receivers = 10;
    final int events = 10;
    //there are 1 producer + 1 consumer each one emitting events
    final int totalEvents = events * receivers;
    final CountDownLatch eventsProcessed = new CountDownLatch(totalEvents);
    //use 1 capacity and wait until each message has been processed
    final DummyReceiverWorker[] dummyReceiverWorkers = new DummyReceiverWorker[receivers];
    final long globalStart = System.currentTimeMillis();
    final long fixedLatency = 100;
    for (int i = 0; i < receivers; i++) {
        dummyReceiverWorkers[i] = new DummyReceiverWorker();
        dummyReceiverWorkers[i].startedEpochMillis = globalStart;
    }
    final Thread roundRobinReceivers = new Thread(() -> {
        for (int i = 0; i < events; i++) {
            for (DummyReceiverWorker worker : dummyReceiverWorkers) {
                worker.recorder.recordValue(fixedLatency);
                eventsProcessed.countDown();
            }
        }
    });
    roundRobinReceivers.start();
    final File reportFolder = tempTestFolder.newFolder("report");
    final WorkerLatencyWriter latencyWriter = new WorkerLatencyWriter(reportFolder, Arrays.asList(dummyReceiverWorkers));
    final Thread writerThread = new Thread(latencyWriter);
    writerThread.setDaemon(true);
    writerThread.start();
    eventsProcessed.await();
    roundRobinReceivers.join();
    writerThread.interrupt();
    writerThread.join();
    final String latencyFileName = "receiverd-latency.hdr";
    final String[] reports = reportFolder.list((dir, name) -> name.equals(latencyFileName));
    Assert.assertArrayEquals(new String[]{latencyFileName}, reports);
    final File reportFile = new File(reportFolder, Objects.requireNonNull(reports)[0]);
    Assert.assertTrue(reportFile.length() > 0);
    final HistogramLogReader histogramLogReader = new HistogramLogReader(reportFile);
    int totalReports = 0;
    while (histogramLogReader.hasNext()) {
        final EncodableHistogram encodableHistogram = histogramLogReader.nextIntervalHistogram();
        if (encodableHistogram instanceof Histogram) {
            final Histogram histogram = (Histogram) encodableHistogram;
            final long totalCount = histogram.getTotalCount();
            Assert.assertEquals("Each histogram must contain the same number of recorded events of each receiver", events, totalCount);
            Assert.assertEquals("Min recorded value must be " + fixedLatency, fixedLatency, histogram.getMinValue());
            Assert.assertEquals("Max recorded value must be " + fixedLatency, fixedLatency, histogram.getMaxValue());
            Assert.assertEquals("Mean recorded value must be " + fixedLatency, (double) fixedLatency, histogram.getMean(), 0d);
        }
        totalReports++;
    }
    Assert.assertEquals("The histogram number must be the same of the receivers", receivers, totalReports);
}
 

public static void main(final String[] args) throws IOException
{
    final Histogram histogram = (Histogram) new HistogramLogReader(args[0]).nextIntervalHistogram();
    final HistogramReporter reporter = new HistogramReporter(0L, null, "");
    reporter.writeReport(histogram, System.out, EnumSet.of(ReportFormat.LONG), args[0]);
}