Adding unit tests

(cherry picked from commit 4a5c383)

Author: reiabreu

external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/metrics2/KafkaOffsetPartitionAndTopicMetrics.java

@@ -47,19 +47,19 @@ public class KafkaOffsetPartitionAndTopicMetrics implements MetricSet {
     private final Supplier<Map<TopicPartition, OffsetManager>> offsetManagerSupplier;
     private final Supplier<Admin> adminSupplier;
     private final Set<TopicPartition> assignment;
-    private final Map<String, KafkaOffsetTopicMetrics> topicMetricsMap;
-
 
     public KafkaOffsetPartitionAndTopicMetrics(Supplier<Map<TopicPartition, OffsetManager>> offsetManagerSupplier, Supplier<Admin> adminSupplier, Set<TopicPartition> assignment) {
         this.offsetManagerSupplier = offsetManagerSupplier;
         this.adminSupplier = adminSupplier;
         this.assignment = assignment;
-        this.topicMetricsMap=new HashMap<>();
+
     }
 
     @Override
     public Map<String, Metric> getMetrics() {
 
+        Map<String, KafkaOffsetTopicMetrics> topicMetricsMap=new HashMap<>();
+
         Map<String, Metric> metrics = new HashMap<>();
 
         for (TopicPartition topicPartition : assignment) {

external/storm-kafka-client/src/test/java/org/apache/storm/kafka/spout/metric2/KafkaOffsetPartitionAndTopicMetricsTest.java

@@ -101,6 +101,23 @@ public void registerPartitionAndTopicMetrics() throws ExecutionException, Interr
         Gauge gBTotal= (Gauge) result.get("topicB/totalSpoutLag");
         assertEquals(gBTotal.getValue(),150L);
 
+        //get the metrics a second time. Values should be the same. In particular, the total values for the topic should not accumulate. Each call to getMetrics should reset the total values.
+
+        result = kafkaOffsetPartitionAndTopicMetrics.getMetrics();
+         g1= (Gauge) result.get("topicA/partition_1/spoutLag");
+         g2= (Gauge) result.get("topicA/partition_2/spoutLag");
+         g3= (Gauge) result.get("topicB/partition_1/spoutLag");
+         g4= (Gauge) result.get("topicB/partition_2/spoutLag");
+        assertEquals(g1.getValue(),10L);
+        assertEquals(g2.getValue(),30L);
+        assertEquals(g3.getValue(),100L);
+        assertEquals(g4.getValue(),50L);
+
+        gATotal= (Gauge) result.get("topicA/totalSpoutLag");
+        assertEquals(gATotal.getValue(),40L);
+        gBTotal= (Gauge) result.get("topicB/totalSpoutLag");
+        assertEquals(gBTotal.getValue(),150L);
+
 
     }
 }