@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); }
@VisibleForTesting public static String convertHistogramToString(Histogram observedLatencyHistogram) { ByteArrayOutputStream baos = new ByteArrayOutputStream(); try (PrintStream stream = new PrintStream(baos, true, Charsets.UTF_8.name())) { HistogramLogWriter histogramLogWriter = new HistogramLogWriter(stream); histogramLogWriter.outputIntervalHistogram(observedLatencyHistogram); return new String(baos.toByteArray(), Charsets.UTF_8); } catch (UnsupportedEncodingException e) { log.error("Exception {} encountered when creating PrintStream; returning empty string", e); return EMPTY_STRING; } }
KafkaExtractorStatsTracker { @VisibleForTesting public static String convertHistogramToString(Histogram observedLatencyHistogram) { ByteArrayOutputStream baos = new ByteArrayOutputStream(); try (PrintStream stream = new PrintStream(baos, true, Charsets.UTF_8.name())) { HistogramLogWriter histogramLogWriter = new HistogramLogWriter(stream); histogramLogWriter.outputIntervalHistogram(observedLatencyHistogram); return new String(baos.toByteArray(), Charsets.UTF_8); } catch (UnsupportedEncodingException e) { log.error("Exception {} encountered when creating PrintStream; returning empty string", e); return EMPTY_STRING; } } }
KafkaExtractorStatsTracker { @VisibleForTesting public static String convertHistogramToString(Histogram observedLatencyHistogram) { ByteArrayOutputStream baos = new ByteArrayOutputStream(); try (PrintStream stream = new PrintStream(baos, true, Charsets.UTF_8.name())) { HistogramLogWriter histogramLogWriter = new HistogramLogWriter(stream); histogramLogWriter.outputIntervalHistogram(observedLatencyHistogram); return new String(baos.toByteArray(), Charsets.UTF_8); } catch (UnsupportedEncodingException e) { log.error("Exception {} encountered when creating PrintStream; returning empty string", e); return EMPTY_STRING; } } KafkaExtractorStatsTracker(WorkUnitState state, List<KafkaPartition> partitions); }
KafkaExtractorStatsTracker { @VisibleForTesting public static String convertHistogramToString(Histogram observedLatencyHistogram) { ByteArrayOutputStream baos = new ByteArrayOutputStream(); try (PrintStream stream = new PrintStream(baos, true, Charsets.UTF_8.name())) { HistogramLogWriter histogramLogWriter = new HistogramLogWriter(stream); histogramLogWriter.outputIntervalHistogram(observedLatencyHistogram); return new String(baos.toByteArray(), Charsets.UTF_8); } catch (UnsupportedEncodingException e) { log.error("Exception {} encountered when creating PrintStream; returning empty string", e); return EMPTY_STRING; } } KafkaExtractorStatsTracker(WorkUnitState state, List<KafkaPartition> partitions); int getErrorPartitionCount(); Long getDecodingErrorCount(int partitionIdx); void onUndecodeableRecord(int partitionIdx); void resetStartFetchEpochTime(int partitionIdx); void onDecodeableRecord(int partitionIdx, long readStartTime, long decodeStartTime, long recordSizeInBytes, long logAppendTimestamp, long recordCreationTimestamp); void onFetchNextMessageBuffer(int partitionIdx, long fetchStartTime); void updateStatisticsForCurrentPartition(int partitionIdx, long readStartTime, long lastSuccessfulRecordHeaderTimestamp); @VisibleForTesting static String convertHistogramToString(Histogram observedLatencyHistogram); void emitTrackingEvents(MetricContext context, Map<KafkaPartition, Map<String, String>> tagsForPartitionsMap); Map<KafkaPartition, Map<String, String>> generateTagsForPartitions(MultiLongWatermark lowWatermark, MultiLongWatermark highWatermark,
      MultiLongWatermark nextWatermark, Map<KafkaPartition, Map<String, String>> additionalTags); long getAvgRecordSize(int partitionIdx); long getMaxIngestionLatency(TimeUnit timeUnit); double getConsumptionRateMBps(); void reset(); }
KafkaExtractorStatsTracker { @VisibleForTesting public static String convertHistogramToString(Histogram observedLatencyHistogram) { ByteArrayOutputStream baos = new ByteArrayOutputStream(); try (PrintStream stream = new PrintStream(baos, true, Charsets.UTF_8.name())) { HistogramLogWriter histogramLogWriter = new HistogramLogWriter(stream); histogramLogWriter.outputIntervalHistogram(observedLatencyHistogram); return new String(baos.toByteArray(), Charsets.UTF_8); } catch (UnsupportedEncodingException e) { log.error("Exception {} encountered when creating PrintStream; returning empty string", e); return EMPTY_STRING; } } KafkaExtractorStatsTracker(WorkUnitState state, List<KafkaPartition> partitions); int getErrorPartitionCount(); Long getDecodingErrorCount(int partitionIdx); void onUndecodeableRecord(int partitionIdx); void resetStartFetchEpochTime(int partitionIdx); void onDecodeableRecord(int partitionIdx, long readStartTime, long decodeStartTime, long recordSizeInBytes, long logAppendTimestamp, long recordCreationTimestamp); void onFetchNextMessageBuffer(int partitionIdx, long fetchStartTime); void updateStatisticsForCurrentPartition(int partitionIdx, long readStartTime, long lastSuccessfulRecordHeaderTimestamp); @VisibleForTesting static String convertHistogramToString(Histogram observedLatencyHistogram); void emitTrackingEvents(MetricContext context, Map<KafkaPartition, Map<String, String>> tagsForPartitionsMap); Map<KafkaPartition, Map<String, String>> generateTagsForPartitions(MultiLongWatermark lowWatermark, MultiLongWatermark highWatermark,
      MultiLongWatermark nextWatermark, Map<KafkaPartition, Map<String, String>> additionalTags); long getAvgRecordSize(int partitionIdx); long getMaxIngestionLatency(TimeUnit timeUnit); double getConsumptionRateMBps(); void reset(); static final String TOPIC; static final String PARTITION; }