@Test public void testConnectorPartitionMetrics() { KafkaBasedConnectorTaskMetrics connectorConsumer1 = new KafkaBasedConnectorTaskMetrics(CLASS_NAME, CONSUMER1_NAME, LOG, false); KafkaBasedConnectorTaskMetrics connectorConsumer2 = new KafkaBasedConnectorTaskMetrics(CLASS_NAME, CONSUMER2_NAME, LOG, false); connectorConsumer1.createPartitionMetrics(); connectorConsumer2.createPartitionMetrics(); Random random = new Random(); long consumer1NumTopics = random.nextInt(1000); long consumer1NumPartitions = consumer1NumTopics * random.nextInt(1024); long consumer2NumTopics = random.nextInt(1000); long consumer2NumPartitions = consumer2NumTopics * random.nextInt(1024); connectorConsumer1.updateNumTopics(consumer1NumTopics); connectorConsumer1.updateNumPartitions(consumer1NumPartitions); connectorConsumer2.updateNumTopics(consumer2NumTopics); connectorConsumer2.updateNumPartitions(consumer2NumPartitions); Assert.assertEquals((long) ((Gauge) _metricsManager.getMetric(CLASS_NAME + DELIMITED_CONSUMER1_NAME + "numTopics")).getValue(), consumer1NumTopics); Assert.assertEquals((long) ((Gauge) _metricsManager.getMetric(CLASS_NAME + DELIMITED_CONSUMER1_NAME + "numPartitions")).getValue(), consumer1NumPartitions); Assert.assertEquals((long) ((Gauge) _metricsManager.getMetric(CLASS_NAME + DELIMITED_CONSUMER2_NAME + "numTopics")).getValue(), consumer2NumTopics); Assert.assertEquals((long) ((Gauge) _metricsManager.getMetric(CLASS_NAME + DELIMITED_CONSUMER2_NAME + "numPartitions")).getValue(), consumer2NumPartitions); Assert.assertEquals((long) ((Gauge) _metricsManager.getMetric(CLASS_NAME + DELIMITED_AGGREGATE_NAME + "numTopics")).getValue(), consumer1NumTopics + consumer2NumTopics); Assert.assertEquals((long) ((Gauge) _metricsManager.getMetric(CLASS_NAME + DELIMITED_AGGREGATE_NAME + "numPartitions")).getValue(), consumer1NumPartitions + consumer2NumPartitions); consumer1NumTopics += random.nextInt(1000); consumer1NumPartitions = consumer1NumTopics * random.nextInt(1024); consumer2NumTopics += random.nextInt(1000); consumer2NumPartitions = consumer2NumTopics * random.nextInt(1024); connectorConsumer1.updateNumTopics(consumer1NumTopics); connectorConsumer1.updateNumPartitions(consumer1NumPartitions); connectorConsumer2.updateNumTopics(consumer2NumTopics); connectorConsumer2.updateNumPartitions(consumer2NumPartitions); Assert.assertEquals((long) ((Gauge) _metricsManager.getMetric(CLASS_NAME + DELIMITED_CONSUMER1_NAME + "numTopics")).getValue(), consumer1NumTopics); Assert.assertEquals((long) ((Gauge) _metricsManager.getMetric(CLASS_NAME + DELIMITED_CONSUMER1_NAME + "numPartitions")).getValue(), consumer1NumPartitions); Assert.assertEquals((long) ((Gauge) _metricsManager.getMetric(CLASS_NAME + DELIMITED_CONSUMER2_NAME + "numTopics")).getValue(), consumer2NumTopics); Assert.assertEquals((long) ((Gauge) _metricsManager.getMetric(CLASS_NAME + DELIMITED_CONSUMER2_NAME + "numPartitions")).getValue(), consumer2NumPartitions); Assert.assertEquals((long) ((Gauge) _metricsManager.getMetric(CLASS_NAME + DELIMITED_AGGREGATE_NAME + "numTopics")).getValue(), consumer1NumTopics + consumer2NumTopics); Assert.assertEquals((long) ((Gauge) _metricsManager.getMetric(CLASS_NAME + DELIMITED_AGGREGATE_NAME + "numPartitions")).getValue(), consumer1NumPartitions + consumer2NumPartitions); }
public void updateNumTopics(long val) { long delta = val - _numTopics.getAndSet(val); AtomicLong aggregatedMetric = AGGREGATED_NUM_TOPICS.get(_className); if (aggregatedMetric != null) { aggregatedMetric.getAndAdd(delta); } }
KafkaBasedConnectorTaskMetrics extends CommonConnectorMetrics { public void updateNumTopics(long val) { long delta = val - _numTopics.getAndSet(val); AtomicLong aggregatedMetric = AGGREGATED_NUM_TOPICS.get(_className); if (aggregatedMetric != null) { aggregatedMetric.getAndAdd(delta); } } }
KafkaBasedConnectorTaskMetrics extends CommonConnectorMetrics { public void updateNumTopics(long val) { long delta = val - _numTopics.getAndSet(val); AtomicLong aggregatedMetric = AGGREGATED_NUM_TOPICS.get(_className); if (aggregatedMetric != null) { aggregatedMetric.getAndAdd(delta); } } KafkaBasedConnectorTaskMetrics(String className, String metricsKey, Logger errorLogger,
      boolean enablePollDurationMillisMetric); }
KafkaBasedConnectorTaskMetrics extends CommonConnectorMetrics { public void updateNumTopics(long val) { long delta = val - _numTopics.getAndSet(val); AtomicLong aggregatedMetric = AGGREGATED_NUM_TOPICS.get(_className); if (aggregatedMetric != null) { aggregatedMetric.getAndAdd(delta); } } KafkaBasedConnectorTaskMetrics(String className, String metricsKey, Logger errorLogger,
      boolean enablePollDurationMillisMetric); @Override void deregisterMetrics(); void updateNumConfigPausedPartitions(long val); void updateNumAutoPausedPartitionsOnError(long val); void updateNumAutoPausedPartitionsOnInFlightMessages(long val); void updateNumAutoPausedPartitionsAwaitingDestTopic(long val); void updateNumTopics(long val); void updatePollDurationMs(long val); static List<BrooklinMetricInfo> getKafkaBasedConnectorTaskSpecificMetrics(String prefix); }
KafkaBasedConnectorTaskMetrics extends CommonConnectorMetrics { public void updateNumTopics(long val) { long delta = val - _numTopics.getAndSet(val); AtomicLong aggregatedMetric = AGGREGATED_NUM_TOPICS.get(_className); if (aggregatedMetric != null) { aggregatedMetric.getAndAdd(delta); } } KafkaBasedConnectorTaskMetrics(String className, String metricsKey, Logger errorLogger,
      boolean enablePollDurationMillisMetric); @Override void deregisterMetrics(); void updateNumConfigPausedPartitions(long val); void updateNumAutoPausedPartitionsOnError(long val); void updateNumAutoPausedPartitionsOnInFlightMessages(long val); void updateNumAutoPausedPartitionsAwaitingDestTopic(long val); void updateNumTopics(long val); void updatePollDurationMs(long val); static List<BrooklinMetricInfo> getKafkaBasedConnectorTaskSpecificMetrics(String prefix); static final String NUM_CONFIG_PAUSED_PARTITIONS; static final String NUM_AUTO_PAUSED_PARTITIONS_ON_ERROR; static final String NUM_AUTO_PAUSED_PARTITIONS_ON_INFLIGHT_MESSAGES; static final String NUM_AUTO_PAUSED_PARTITIONS_WAITING_FOR_DEST_TOPIC; static final String NUM_TOPICS; static final String POLL_DURATION_MS; }