@Test public void testPublishMetrics() { KafkaProducerConfig config = new KafkaProducerConfig("/var/serverset/discovery.kafka.prod", Arrays.asList("localhost:9092"), "-1"); KafkaProducerManager.getInstance().getProducers().clear(); KafkaProducer<byte[], byte[]> producer = KafkaProducerManager.getProducer(config); KafkaProducerMetricsMonitor monitor = new KafkaProducerMetricsMonitor(); monitor.publishKafkaProducerMetricsToOstrich(); producer.close(); for (String metricName : KafkaProducerMetricsMonitor.PRODUCER_METRICS_ALLOWLIST) { Object gauge = Stats.getGauge("kafkaproducer." + metricName + " cluster=kafka").get(); assertNotNull(gauge); } }
@SuppressWarnings({ "deprecation" }) protected void publishKafkaProducerMetricsToOstrich() { Map<KafkaProducerConfig, KafkaProducer<byte[], byte[]>> producers = KafkaProducerManager .getInstance().getProducers(); for (Entry<KafkaProducerConfig, KafkaProducer<byte[], byte[]>> kafkaProducerEntry : producers .entrySet()) { KafkaProducerConfig key = kafkaProducerEntry.getKey(); String signature = convertSignatureToTag(key); Map<MetricName, ? extends Metric> metrics = kafkaProducerEntry.getValue().metrics(); for (Entry<MetricName, ? extends Metric> entry : metrics.entrySet()) { if (PRODUCER_METRICS_ALLOWLIST.contains(entry.getKey().name())) { OpenTsdbMetricConverter.gauge("kafkaproducer." + entry.getKey().name(), entry.getValue().value(), "cluster=" + signature); } } } }
KafkaProducerMetricsMonitor implements Runnable { @SuppressWarnings({ "deprecation" }) protected void publishKafkaProducerMetricsToOstrich() { Map<KafkaProducerConfig, KafkaProducer<byte[], byte[]>> producers = KafkaProducerManager .getInstance().getProducers(); for (Entry<KafkaProducerConfig, KafkaProducer<byte[], byte[]>> kafkaProducerEntry : producers .entrySet()) { KafkaProducerConfig key = kafkaProducerEntry.getKey(); String signature = convertSignatureToTag(key); Map<MetricName, ? extends Metric> metrics = kafkaProducerEntry.getValue().metrics(); for (Entry<MetricName, ? extends Metric> entry : metrics.entrySet()) { if (PRODUCER_METRICS_ALLOWLIST.contains(entry.getKey().name())) { OpenTsdbMetricConverter.gauge("kafkaproducer." + entry.getKey().name(), entry.getValue().value(), "cluster=" + signature); } } } } }
KafkaProducerMetricsMonitor implements Runnable { @SuppressWarnings({ "deprecation" }) protected void publishKafkaProducerMetricsToOstrich() { Map<KafkaProducerConfig, KafkaProducer<byte[], byte[]>> producers = KafkaProducerManager .getInstance().getProducers(); for (Entry<KafkaProducerConfig, KafkaProducer<byte[], byte[]>> kafkaProducerEntry : producers .entrySet()) { KafkaProducerConfig key = kafkaProducerEntry.getKey(); String signature = convertSignatureToTag(key); Map<MetricName, ? extends Metric> metrics = kafkaProducerEntry.getValue().metrics(); for (Entry<MetricName, ? extends Metric> entry : metrics.entrySet()) { if (PRODUCER_METRICS_ALLOWLIST.contains(entry.getKey().name())) { OpenTsdbMetricConverter.gauge("kafkaproducer." + entry.getKey().name(), entry.getValue().value(), "cluster=" + signature); } } } }  }
KafkaProducerMetricsMonitor implements Runnable { @SuppressWarnings({ "deprecation" }) protected void publishKafkaProducerMetricsToOstrich() { Map<KafkaProducerConfig, KafkaProducer<byte[], byte[]>> producers = KafkaProducerManager .getInstance().getProducers(); for (Entry<KafkaProducerConfig, KafkaProducer<byte[], byte[]>> kafkaProducerEntry : producers .entrySet()) { KafkaProducerConfig key = kafkaProducerEntry.getKey(); String signature = convertSignatureToTag(key); Map<MetricName, ? extends Metric> metrics = kafkaProducerEntry.getValue().metrics(); for (Entry<MetricName, ? extends Metric> entry : metrics.entrySet()) { if (PRODUCER_METRICS_ALLOWLIST.contains(entry.getKey().name())) { OpenTsdbMetricConverter.gauge("kafkaproducer." + entry.getKey().name(), entry.getValue().value(), "cluster=" + signature); } } } }  @Override void run(); static String convertSignatureToTag(KafkaProducerConfig key); }
KafkaProducerMetricsMonitor implements Runnable { @SuppressWarnings({ "deprecation" }) protected void publishKafkaProducerMetricsToOstrich() { Map<KafkaProducerConfig, KafkaProducer<byte[], byte[]>> producers = KafkaProducerManager .getInstance().getProducers(); for (Entry<KafkaProducerConfig, KafkaProducer<byte[], byte[]>> kafkaProducerEntry : producers .entrySet()) { KafkaProducerConfig key = kafkaProducerEntry.getKey(); String signature = convertSignatureToTag(key); Map<MetricName, ? extends Metric> metrics = kafkaProducerEntry.getValue().metrics(); for (Entry<MetricName, ? extends Metric> entry : metrics.entrySet()) { if (PRODUCER_METRICS_ALLOWLIST.contains(entry.getKey().name())) { OpenTsdbMetricConverter.gauge("kafkaproducer." + entry.getKey().name(), entry.getValue().value(), "cluster=" + signature); } } } }  @Override void run(); static String convertSignatureToTag(KafkaProducerConfig key); static final Set<String> PRODUCER_METRICS_ALLOWLIST; }