CommonConnectorMetrics { public void deregisterMetrics() { _metricsList.forEach(BrooklinMetrics::deregister); } CommonConnectorMetrics(String className, String key, Logger errorLogger); void createEventProcessingMetrics(); void createPollMetrics(); void createPartitionMetrics(); void deregisterMetrics(); void updateEventsProcessedRate(long val); void updateBytesProcessedRate(long val); void updateErrorRate(long val); void updateErrorRate(long val, String message, Exception e); void updateAggregatedErrorRate(long val, String message, Exception e); void updateProcessingAboveThreshold(long val); void updateLastEventReceivedTime(Instant val); void updateNumPolls(long val); void updateEventCountsPerPoll(long val); void updateClientPollOverTimeout(long val); void updateClientPollIntervalOverSessionTimeout(long val); void resetStuckPartitions(); void updateStuckPartitions(long val); void updateNumPartitions(long val); void updateRebalanceRate(long val); static List<BrooklinMetricInfo> getEventProcessingMetrics(String prefix); static List<BrooklinMetricInfo> getEventPollMetrics(String prefix); static List<BrooklinMetricInfo> getPartitionSpecificMetrics(String prefix); static final String AGGREGATE; }
@Test public void testDeregisterMetrics() { CommonConnectorMetrics metrics1 = new CommonConnectorMetrics(CLASS_NAME, CONSUMER1_NAME, LOG); CommonConnectorMetrics metrics2 = new CommonConnectorMetrics(CLASS_NAME, CONSUMER2_NAME, LOG); metrics1.createPartitionMetrics(); metrics2.createPartitionMetrics(); metrics1.updateNumPartitions(15); metrics2.updateNumPartitions(10); Assert.assertEquals((long) ((Gauge) _metricsManager.getMetric(CLASS_NAME + DELIMITED_AGGREGATE + CommonConnectorMetrics.PartitionMetrics.NUM_PARTITIONS)).getValue(), 25); metrics1.updateStuckPartitions(10); metrics2.updateStuckPartitions(15); Assert.assertEquals((long) ((Gauge) _metricsManager.getMetric(CLASS_NAME + DELIMITED_AGGREGATE + CommonConnectorMetrics.PartitionMetrics.STUCK_PARTITIONS)).getValue(), 25); Assert.assertNotNull(_metricsManager.getMetric(String.join(".", CLASS_NAME, "aggregate", "stuckPartitions"))); Assert.assertNotNull(_metricsManager.getMetric(String.join(".", CLASS_NAME, CONSUMER1_NAME, "stuckPartitions"))); Assert.assertNotNull(_metricsManager.getMetric(String.join(".", CLASS_NAME, CONSUMER2_NAME, "stuckPartitions"))); metrics1.deregisterMetrics(); Assert.assertNotNull(_metricsManager.getMetric(String.join(".", CLASS_NAME, "aggregate", "stuckPartitions"))); Assert.assertNull(_metricsManager.getMetric(String.join(".", CLASS_NAME, CONSUMER1_NAME, "stuckPartitions"))); Assert.assertEquals((long) ((Gauge) _metricsManager.getMetric(CLASS_NAME + DELIMITED_AGGREGATE + CommonConnectorMetrics.PartitionMetrics.NUM_PARTITIONS)).getValue(), 10); Assert.assertEquals((long) ((Gauge) _metricsManager.getMetric(CLASS_NAME + DELIMITED_AGGREGATE + CommonConnectorMetrics.PartitionMetrics.STUCK_PARTITIONS)).getValue(), 15); metrics2.deregisterMetrics(); Assert.assertNull(_metricsManager.getMetric(String.join(".", CLASS_NAME, "aggregate", "stuckPartitions"))); Assert.assertNull(_metricsManager.getMetric(String.join(".", CLASS_NAME, CONSUMER2_NAME, "stuckPartitions"))); CommonConnectorMetrics metrics3 = new CommonConnectorMetrics(CLASS_NAME, CONSUMER3_NAME, LOG); metrics3.createPartitionMetrics(); Assert.assertNotNull(_metricsManager.getMetric(String.join(".", CLASS_NAME, "aggregate", "stuckPartitions"))); Assert.assertNotNull(_metricsManager.getMetric(String.join(".", CLASS_NAME, CONSUMER3_NAME, "stuckPartitions"))); metrics3.deregisterMetrics(); Assert.assertNull(_metricsManager.getMetric(String.join(".", CLASS_NAME, "aggregate", "stuckPartitions"))); Assert.assertNull(_metricsManager.getMetric(String.join(".", CLASS_NAME, CONSUMER3_NAME, "stuckPartitions"))); }