@Test public void testOnUndecodeableRecord() { Assert.assertEquals(this.extractorStatsTracker.getErrorPartitionCount(), 0); Assert.assertEquals(this.extractorStatsTracker.getDecodingErrorCount(0).longValue(), -1); Assert.assertEquals(this.extractorStatsTracker.getDecodingErrorCount(0).longValue(), -1); this.extractorStatsTracker.onUndecodeableRecord(0); Assert.assertEquals(this.extractorStatsTracker.getDecodingErrorCount(0).longValue(), 1); Assert.assertEquals(this.extractorStatsTracker.getDecodingErrorCount(1).longValue(), -1); Assert.assertEquals(this.extractorStatsTracker.getErrorPartitionCount(), 1); this.extractorStatsTracker.onUndecodeableRecord(0); Assert.assertEquals(this.extractorStatsTracker.getDecodingErrorCount(0).longValue(), 2); Assert.assertEquals(this.extractorStatsTracker.getDecodingErrorCount(1).longValue(), -1); Assert.assertEquals(this.extractorStatsTracker.getErrorPartitionCount(), 1); }
public void onUndecodeableRecord(int partitionIdx) { this.errorPartitions.add(partitionIdx); this.undecodableMessageCount++; incrementErrorCount(partitionIdx); }
KafkaExtractorStatsTracker { public void onUndecodeableRecord(int partitionIdx) { this.errorPartitions.add(partitionIdx); this.undecodableMessageCount++; incrementErrorCount(partitionIdx); } }
KafkaExtractorStatsTracker { public void onUndecodeableRecord(int partitionIdx) { this.errorPartitions.add(partitionIdx); this.undecodableMessageCount++; incrementErrorCount(partitionIdx); } KafkaExtractorStatsTracker(WorkUnitState state, List<KafkaPartition> partitions); }
KafkaExtractorStatsTracker { public void onUndecodeableRecord(int partitionIdx) { this.errorPartitions.add(partitionIdx); this.undecodableMessageCount++; incrementErrorCount(partitionIdx); } 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 onUndecodeableRecord(int partitionIdx) { this.errorPartitions.add(partitionIdx); this.undecodableMessageCount++; incrementErrorCount(partitionIdx); } 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; }