Add metric to track number of kafka producers, this will help us

determine if we need to tune the producer buffers to lower the memory
footprint

Author: ambud

singer/src/main/java/com/pinterest/singer/common/SingerMetrics.java

@@ -75,6 +75,7 @@ public class SingerMetrics {
   public static final String KAFKA_LATENCY = SINGER_WRITER + "max_kafka_batch_write_latency";
   public static final String NUM_COMMITED_TRANSACTIONS = SINGER_WRITER + "num_committed_transactions";
   public static final String NUM_ABORTED_TRANSACTIONS = SINGER_WRITER + "num_aborted_transactions";
+  public static final String NUM_KAFKA_PRODUCERS = SINGER_WRITER + "num_kafka_producers";
 
   public static final String KUBE_PREFIX           = SINGER_PREIX + "kube.";
   public static final String KUBE_API_ERROR        = KUBE_PREFIX  + "api_error";

singer/src/main/java/com/pinterest/singer/writer/KafkaProducerManager.java

@@ -15,6 +15,8 @@
  */
 package com.pinterest.singer.writer;
 
+import com.pinterest.singer.common.SingerMetrics;
+import com.pinterest.singer.metrics.OpenTsdbMetricConverter;
 import com.pinterest.singer.thrift.configuration.KafkaProducerConfig;
 import com.pinterest.singer.utils.KafkaUtils;
 
@@ -81,6 +83,8 @@ private KafkaProducer<byte[], byte[]> getProducerInternal(KafkaProducerConfig co
       if (result != null && result != producer) {
         producer.close();
       }
+      // log metrics for no.of kafka producers currently in the cache
+      OpenTsdbMetricConverter.addMetric(SingerMetrics.NUM_KAFKA_PRODUCERS, producers.size());
     }
     result = producers.get(config);
     return result;
@@ -106,6 +110,8 @@ private boolean resetProducerInternal(KafkaProducerConfig config) {
       if (!retval) {
         newProducer.close();
       }
+      // log metrics for no.of kafka producers currently in the cache
+      OpenTsdbMetricConverter.addMetric(SingerMetrics.NUM_KAFKA_PRODUCERS, producers.size());
     }
     return retval;
   }