@Test (dependsOnMethods = "testUpdateStatisticsForCurrentPartition") public void testGetAvgRecordSize() { Assert.assertEquals(this.extractorStatsTracker.getAvgRecordSize(0), 100); Assert.assertEquals(this.extractorStatsTracker.getAvgRecordSize(1), 100); this.extractorStatsTracker.reset(); Assert.assertEquals(this.extractorStatsTracker.getAvgRecordSize(0), 0); long readStartTime = System.nanoTime(); long decodeStartTime = readStartTime + 1; long currentTimeMillis = System.currentTimeMillis(); long logAppendTimestamp = currentTimeMillis - 10; long recordCreationTimestamp = currentTimeMillis - 20; this.extractorStatsTracker.onDecodeableRecord(1, readStartTime, decodeStartTime, 150, logAppendTimestamp, recordCreationTimestamp); Assert.assertEquals(this.extractorStatsTracker.getAvgRecordSize(1), 150); }
public long getAvgRecordSize(int partitionIdx) { ExtractorStats stats = this.statsMap.getOrDefault(this.partitions.get(partitionIdx), null); if (stats != null) { if (stats.getAvgRecordSize() != 0) { return stats.getAvgRecordSize(); } else { if (stats.getProcessedRecordCount() != 0) { return stats.getPartitionTotalSize() / stats.getProcessedRecordCount(); } } } return 0; }
KafkaExtractorStatsTracker { public long getAvgRecordSize(int partitionIdx) { ExtractorStats stats = this.statsMap.getOrDefault(this.partitions.get(partitionIdx), null); if (stats != null) { if (stats.getAvgRecordSize() != 0) { return stats.getAvgRecordSize(); } else { if (stats.getProcessedRecordCount() != 0) { return stats.getPartitionTotalSize() / stats.getProcessedRecordCount(); } } } return 0; } }
KafkaExtractorStatsTracker { public long getAvgRecordSize(int partitionIdx) { ExtractorStats stats = this.statsMap.getOrDefault(this.partitions.get(partitionIdx), null); if (stats != null) { if (stats.getAvgRecordSize() != 0) { return stats.getAvgRecordSize(); } else { if (stats.getProcessedRecordCount() != 0) { return stats.getPartitionTotalSize() / stats.getProcessedRecordCount(); } } } return 0; } KafkaExtractorStatsTracker(WorkUnitState state, List<KafkaPartition> partitions); }
KafkaExtractorStatsTracker { public long getAvgRecordSize(int partitionIdx) { ExtractorStats stats = this.statsMap.getOrDefault(this.partitions.get(partitionIdx), null); if (stats != null) { if (stats.getAvgRecordSize() != 0) { return stats.getAvgRecordSize(); } else { if (stats.getProcessedRecordCount() != 0) { return stats.getPartitionTotalSize() / stats.getProcessedRecordCount(); } } } return 0; } 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 long getAvgRecordSize(int partitionIdx) { ExtractorStats stats = this.statsMap.getOrDefault(this.partitions.get(partitionIdx), null); if (stats != null) { if (stats.getAvgRecordSize() != 0) { return stats.getAvgRecordSize(); } else { if (stats.getProcessedRecordCount() != 0) { return stats.getPartitionTotalSize() / stats.getProcessedRecordCount(); } } } return 0; } 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; }