@Test public void testOnDecodeableRecord() throws InterruptedException { this.extractorStatsTracker.reset(); long readStartTime = System.nanoTime(); Thread.sleep(1); long decodeStartTime = System.nanoTime(); long currentTimeMillis = System.currentTimeMillis(); long logAppendTimestamp = currentTimeMillis - TimeUnit.MINUTES.toMillis(15); long recordCreationTimestamp = currentTimeMillis - TimeUnit.MINUTES.toMillis(16); Assert.assertEquals(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getProcessedRecordCount(), 0); Assert.assertEquals(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getPartitionTotalSize(), 0); Assert.assertTrue(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getDecodeRecordTime() == 0); Assert.assertTrue(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getReadRecordTime() == 0); Assert.assertEquals(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getSlaMissedRecordCount(), -1); Assert.assertEquals(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getMinLogAppendTime(), -1); Assert.assertEquals(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getMaxLogAppendTime(), -1); Assert.assertEquals(this.extractorStatsTracker.getObservedLatencyHistogram().getTotalCount(), 0); this.extractorStatsTracker.onDecodeableRecord(0, readStartTime, decodeStartTime, 100, logAppendTimestamp, recordCreationTimestamp); Assert.assertEquals(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getProcessedRecordCount(), 1); Assert.assertEquals(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getPartitionTotalSize(), 100); Assert.assertTrue(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getDecodeRecordTime() > 0); Assert.assertTrue(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getReadRecordTime() > 0); Assert.assertEquals(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getSlaMissedRecordCount(), 1); Assert.assertEquals(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getMinLogAppendTime(), logAppendTimestamp); Assert.assertEquals(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getMaxLogAppendTime(), logAppendTimestamp); Assert.assertEquals(this.extractorStatsTracker.getObservedLatencyHistogram().getTotalCount(), 1); readStartTime = System.nanoTime(); Thread.sleep(1); decodeStartTime = System.nanoTime(); long previousLogAppendTimestamp = logAppendTimestamp; currentTimeMillis = System.currentTimeMillis(); logAppendTimestamp = currentTimeMillis - 10; recordCreationTimestamp = currentTimeMillis - 20; long previousDecodeRecordTime = this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getDecodeRecordTime(); long previousReadRecordTime = this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getReadRecordTime(); this.extractorStatsTracker.onDecodeableRecord(0, readStartTime, decodeStartTime, 100, logAppendTimestamp, recordCreationTimestamp); Assert.assertEquals(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getProcessedRecordCount(), 2); Assert.assertEquals(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getPartitionTotalSize(), 200); Assert.assertTrue(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getDecodeRecordTime() > previousDecodeRecordTime); Assert.assertTrue(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getReadRecordTime() > previousReadRecordTime); Assert.assertEquals(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getSlaMissedRecordCount(), 1); Assert.assertEquals(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getMinLogAppendTime(), previousLogAppendTimestamp); Assert.assertEquals(this.extractorStatsTracker.getStatsMap().get(kafkaPartitions.get(0)).getMaxLogAppendTime(), logAppendTimestamp); Assert.assertEquals(this.extractorStatsTracker.getObservedLatencyHistogram().getTotalCount(), 2); }
public void onDecodeableRecord(int partitionIdx, long readStartTime, long decodeStartTime, long recordSizeInBytes, long logAppendTimestamp, long recordCreationTimestamp) { this.statsMap.computeIfPresent(this.partitions.get(partitionIdx), (k, v) -> { long currentTime = System.nanoTime(); v.processedRecordCount++; v.partitionTotalSize += recordSizeInBytes; v.decodeRecordTime += currentTime - decodeStartTime; v.readRecordTime += currentTime - readStartTime; if (this.observedLatencyHistogram != null && recordCreationTimestamp > 0) { long observedLatency = System.currentTimeMillis() - recordCreationTimestamp; if (observedLatency < this.maxPossibleLatency) { this.observedLatencyHistogram.recordValue(observedLatency); } } if (this.isSlaConfigured) { if (v.slaMissedRecordCount < 0) { v.slaMissedRecordCount = 0; v.minLogAppendTime = logAppendTimestamp; v.maxLogAppendTime = logAppendTimestamp; } else { if (logAppendTimestamp < v.minLogAppendTime) { v.minLogAppendTime = logAppendTimestamp; } if (logAppendTimestamp > v.maxLogAppendTime) { v.maxLogAppendTime = logAppendTimestamp; } } if (logAppendTimestamp > 0 && (System.currentTimeMillis() - logAppendTimestamp > recordLevelSlaMillis)) { v.slaMissedRecordCount++; } } return v; }); }
KafkaExtractorStatsTracker { public void onDecodeableRecord(int partitionIdx, long readStartTime, long decodeStartTime, long recordSizeInBytes, long logAppendTimestamp, long recordCreationTimestamp) { this.statsMap.computeIfPresent(this.partitions.get(partitionIdx), (k, v) -> { long currentTime = System.nanoTime(); v.processedRecordCount++; v.partitionTotalSize += recordSizeInBytes; v.decodeRecordTime += currentTime - decodeStartTime; v.readRecordTime += currentTime - readStartTime; if (this.observedLatencyHistogram != null && recordCreationTimestamp > 0) { long observedLatency = System.currentTimeMillis() - recordCreationTimestamp; if (observedLatency < this.maxPossibleLatency) { this.observedLatencyHistogram.recordValue(observedLatency); } } if (this.isSlaConfigured) { if (v.slaMissedRecordCount < 0) { v.slaMissedRecordCount = 0; v.minLogAppendTime = logAppendTimestamp; v.maxLogAppendTime = logAppendTimestamp; } else { if (logAppendTimestamp < v.minLogAppendTime) { v.minLogAppendTime = logAppendTimestamp; } if (logAppendTimestamp > v.maxLogAppendTime) { v.maxLogAppendTime = logAppendTimestamp; } } if (logAppendTimestamp > 0 && (System.currentTimeMillis() - logAppendTimestamp > recordLevelSlaMillis)) { v.slaMissedRecordCount++; } } return v; }); } }
KafkaExtractorStatsTracker { public void onDecodeableRecord(int partitionIdx, long readStartTime, long decodeStartTime, long recordSizeInBytes, long logAppendTimestamp, long recordCreationTimestamp) { this.statsMap.computeIfPresent(this.partitions.get(partitionIdx), (k, v) -> { long currentTime = System.nanoTime(); v.processedRecordCount++; v.partitionTotalSize += recordSizeInBytes; v.decodeRecordTime += currentTime - decodeStartTime; v.readRecordTime += currentTime - readStartTime; if (this.observedLatencyHistogram != null && recordCreationTimestamp > 0) { long observedLatency = System.currentTimeMillis() - recordCreationTimestamp; if (observedLatency < this.maxPossibleLatency) { this.observedLatencyHistogram.recordValue(observedLatency); } } if (this.isSlaConfigured) { if (v.slaMissedRecordCount < 0) { v.slaMissedRecordCount = 0; v.minLogAppendTime = logAppendTimestamp; v.maxLogAppendTime = logAppendTimestamp; } else { if (logAppendTimestamp < v.minLogAppendTime) { v.minLogAppendTime = logAppendTimestamp; } if (logAppendTimestamp > v.maxLogAppendTime) { v.maxLogAppendTime = logAppendTimestamp; } } if (logAppendTimestamp > 0 && (System.currentTimeMillis() - logAppendTimestamp > recordLevelSlaMillis)) { v.slaMissedRecordCount++; } } return v; }); } KafkaExtractorStatsTracker(WorkUnitState state, List<KafkaPartition> partitions); }
KafkaExtractorStatsTracker { public void onDecodeableRecord(int partitionIdx, long readStartTime, long decodeStartTime, long recordSizeInBytes, long logAppendTimestamp, long recordCreationTimestamp) { this.statsMap.computeIfPresent(this.partitions.get(partitionIdx), (k, v) -> { long currentTime = System.nanoTime(); v.processedRecordCount++; v.partitionTotalSize += recordSizeInBytes; v.decodeRecordTime += currentTime - decodeStartTime; v.readRecordTime += currentTime - readStartTime; if (this.observedLatencyHistogram != null && recordCreationTimestamp > 0) { long observedLatency = System.currentTimeMillis() - recordCreationTimestamp; if (observedLatency < this.maxPossibleLatency) { this.observedLatencyHistogram.recordValue(observedLatency); } } if (this.isSlaConfigured) { if (v.slaMissedRecordCount < 0) { v.slaMissedRecordCount = 0; v.minLogAppendTime = logAppendTimestamp; v.maxLogAppendTime = logAppendTimestamp; } else { if (logAppendTimestamp < v.minLogAppendTime) { v.minLogAppendTime = logAppendTimestamp; } if (logAppendTimestamp > v.maxLogAppendTime) { v.maxLogAppendTime = logAppendTimestamp; } } if (logAppendTimestamp > 0 && (System.currentTimeMillis() - logAppendTimestamp > recordLevelSlaMillis)) { v.slaMissedRecordCount++; } } return v; }); } 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 { public void onDecodeableRecord(int partitionIdx, long readStartTime, long decodeStartTime, long recordSizeInBytes, long logAppendTimestamp, long recordCreationTimestamp) { this.statsMap.computeIfPresent(this.partitions.get(partitionIdx), (k, v) -> { long currentTime = System.nanoTime(); v.processedRecordCount++; v.partitionTotalSize += recordSizeInBytes; v.decodeRecordTime += currentTime - decodeStartTime; v.readRecordTime += currentTime - readStartTime; if (this.observedLatencyHistogram != null && recordCreationTimestamp > 0) { long observedLatency = System.currentTimeMillis() - recordCreationTimestamp; if (observedLatency < this.maxPossibleLatency) { this.observedLatencyHistogram.recordValue(observedLatency); } } if (this.isSlaConfigured) { if (v.slaMissedRecordCount < 0) { v.slaMissedRecordCount = 0; v.minLogAppendTime = logAppendTimestamp; v.maxLogAppendTime = logAppendTimestamp; } else { if (logAppendTimestamp < v.minLogAppendTime) { v.minLogAppendTime = logAppendTimestamp; } if (logAppendTimestamp > v.maxLogAppendTime) { v.maxLogAppendTime = logAppendTimestamp; } } if (logAppendTimestamp > 0 && (System.currentTimeMillis() - logAppendTimestamp > recordLevelSlaMillis)) { v.slaMissedRecordCount++; } } return v; }); } 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; }