@Test public void testGenerateTagsForPartitions() throws Exception { MultiLongWatermark lowWatermark = new MultiLongWatermark(Arrays.asList(new Long(10), new Long(20))); MultiLongWatermark highWatermark = new MultiLongWatermark(Arrays.asList(new Long(20), new Long(30))); MultiLongWatermark nextWatermark = new MultiLongWatermark(Arrays.asList(new Long(15), new Long(25))); Map<KafkaPartition, Map<String, String>> addtionalTags = ImmutableMap.of(PARTITION0, ImmutableMap.of("testKey", "testValue")); this.workUnitState.removeProp(KafkaUtils.getPartitionPropName(KafkaSource.START_FETCH_EPOCH_TIME, 0)); this.workUnitState.removeProp(KafkaUtils.getPartitionPropName(KafkaSource.STOP_FETCH_EPOCH_TIME, 0)); KafkaUtils.setPartitionAvgRecordMillis(this.workUnitState, PARTITION0, 0); KafkaExtractorStatsTracker.ExtractorStats extractorStats = this.extractorStatsTracker.getStatsMap() .get(kafkaPartitions.get(0)); extractorStats.setStartFetchEpochTime(1000); extractorStats.setStopFetchEpochTime(10000); extractorStats.setAvgMillisPerRecord(10.1); Map<KafkaPartition, Map<String, String>> result = extractorStatsTracker.generateTagsForPartitions(lowWatermark, highWatermark, nextWatermark, addtionalTags); Assert.assertEquals(this.workUnitState.getPropAsLong( KafkaUtils.getPartitionPropName(KafkaSource.START_FETCH_EPOCH_TIME, 0)), extractorStats.getStartFetchEpochTime()); Assert.assertEquals(this.workUnitState.getPropAsLong( KafkaUtils.getPartitionPropName(KafkaSource.STOP_FETCH_EPOCH_TIME, 0)), extractorStats.getStopFetchEpochTime()); Assert.assertEquals(KafkaUtils.getPartitionAvgRecordMillis(this.workUnitState, PARTITION0), extractorStats.getAvgMillisPerRecord()); extractorStats.setStartFetchEpochTime(0); extractorStats.setStopFetchEpochTime(0); extractorStats.setAvgMillisPerRecord(-1); Assert.assertTrue(result.get(PARTITION0).containsKey("testKey")); Assert.assertEquals(result.get(PARTITION0).get("testKey"), "testValue"); Assert.assertFalse(result.get(PARTITION1).containsKey("testKey")); }
public Map<KafkaPartition, Map<String, String>> generateTagsForPartitions(MultiLongWatermark lowWatermark, MultiLongWatermark highWatermark, MultiLongWatermark nextWatermark, Map<KafkaPartition, Map<String, String>> additionalTags) { Map<KafkaPartition, Map<String, String>> tagsForPartitionsMap = Maps.newHashMap(); for (int i = 0; i < this.partitions.size(); i++) { KafkaPartition partitionKey = this.partitions.get(i); log.info(String.format("Actual high watermark for partition %s=%d, expected=%d", this.partitions.get(i), nextWatermark.get(i), highWatermark.get(i))); tagsForPartitionsMap .put(this.partitions.get(i), createTagsForPartition(i, lowWatermark, highWatermark, nextWatermark)); if (additionalTags.containsKey(partitionKey)) { tagsForPartitionsMap.get(partitionKey).putAll(additionalTags.get(partitionKey)); } } return tagsForPartitionsMap; }
KafkaExtractorStatsTracker { public Map<KafkaPartition, Map<String, String>> generateTagsForPartitions(MultiLongWatermark lowWatermark, MultiLongWatermark highWatermark, MultiLongWatermark nextWatermark, Map<KafkaPartition, Map<String, String>> additionalTags) { Map<KafkaPartition, Map<String, String>> tagsForPartitionsMap = Maps.newHashMap(); for (int i = 0; i < this.partitions.size(); i++) { KafkaPartition partitionKey = this.partitions.get(i); log.info(String.format("Actual high watermark for partition %s=%d, expected=%d", this.partitions.get(i), nextWatermark.get(i), highWatermark.get(i))); tagsForPartitionsMap .put(this.partitions.get(i), createTagsForPartition(i, lowWatermark, highWatermark, nextWatermark)); if (additionalTags.containsKey(partitionKey)) { tagsForPartitionsMap.get(partitionKey).putAll(additionalTags.get(partitionKey)); } } return tagsForPartitionsMap; } }
KafkaExtractorStatsTracker { public Map<KafkaPartition, Map<String, String>> generateTagsForPartitions(MultiLongWatermark lowWatermark, MultiLongWatermark highWatermark, MultiLongWatermark nextWatermark, Map<KafkaPartition, Map<String, String>> additionalTags) { Map<KafkaPartition, Map<String, String>> tagsForPartitionsMap = Maps.newHashMap(); for (int i = 0; i < this.partitions.size(); i++) { KafkaPartition partitionKey = this.partitions.get(i); log.info(String.format("Actual high watermark for partition %s=%d, expected=%d", this.partitions.get(i), nextWatermark.get(i), highWatermark.get(i))); tagsForPartitionsMap .put(this.partitions.get(i), createTagsForPartition(i, lowWatermark, highWatermark, nextWatermark)); if (additionalTags.containsKey(partitionKey)) { tagsForPartitionsMap.get(partitionKey).putAll(additionalTags.get(partitionKey)); } } return tagsForPartitionsMap; } KafkaExtractorStatsTracker(WorkUnitState state, List<KafkaPartition> partitions); }
KafkaExtractorStatsTracker { public Map<KafkaPartition, Map<String, String>> generateTagsForPartitions(MultiLongWatermark lowWatermark, MultiLongWatermark highWatermark, MultiLongWatermark nextWatermark, Map<KafkaPartition, Map<String, String>> additionalTags) { Map<KafkaPartition, Map<String, String>> tagsForPartitionsMap = Maps.newHashMap(); for (int i = 0; i < this.partitions.size(); i++) { KafkaPartition partitionKey = this.partitions.get(i); log.info(String.format("Actual high watermark for partition %s=%d, expected=%d", this.partitions.get(i), nextWatermark.get(i), highWatermark.get(i))); tagsForPartitionsMap .put(this.partitions.get(i), createTagsForPartition(i, lowWatermark, highWatermark, nextWatermark)); if (additionalTags.containsKey(partitionKey)) { tagsForPartitionsMap.get(partitionKey).putAll(additionalTags.get(partitionKey)); } } return tagsForPartitionsMap; } 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 Map<KafkaPartition, Map<String, String>> generateTagsForPartitions(MultiLongWatermark lowWatermark, MultiLongWatermark highWatermark, MultiLongWatermark nextWatermark, Map<KafkaPartition, Map<String, String>> additionalTags) { Map<KafkaPartition, Map<String, String>> tagsForPartitionsMap = Maps.newHashMap(); for (int i = 0; i < this.partitions.size(); i++) { KafkaPartition partitionKey = this.partitions.get(i); log.info(String.format("Actual high watermark for partition %s=%d, expected=%d", this.partitions.get(i), nextWatermark.get(i), highWatermark.get(i))); tagsForPartitionsMap .put(this.partitions.get(i), createTagsForPartition(i, lowWatermark, highWatermark, nextWatermark)); if (additionalTags.containsKey(partitionKey)) { tagsForPartitionsMap.get(partitionKey).putAll(additionalTags.get(partitionKey)); } } return tagsForPartitionsMap; } 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; }