Add partition offset metric to each topic partition

Before this commit offset consumer was able to fetch offset just from one partition.
This commit solve this.

Change-Id: I2c2c300219e43ab422b237094ad775ca8795169e
Issue-ID: DCAEGEN2-1783
Signed-off-by: kjaniak <kornel.janiak@nokia.com>

2 files changed, 26 insertions, 9 deletions

diff --git a/sources/hv-collector-kafka-consumer/src/test/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/impl/OffsetKafkaConsumerTest.kt b/sources/hv-collector-kafka-consumer/src/test/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/impl/OffsetKafkaConsumerTest.kt
index 3bb9ebae..26616f1c 100644
--- a/sources/hv-collector-kafka-consumer/src/test/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/impl/OffsetKafkaConsumerTest.kt
+++ b/sources/hv-collector-kafka-consumer/src/test/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/impl/OffsetKafkaConsumerTest.kt
@@ -72,7 +72,7 @@ object OffsetKafkaConsumerTest : Spek({
             }
         }
 
-        given("two topics with partition") {
+        given("two topics with one partition each") {
             val topics = setOf(topicName1, topicName2)
             val kafkaSource = OffsetKafkaConsumer(mockedKafkaConsumer, topics, mockedMetrics, testDispatcher)
 
@@ -147,7 +147,7 @@ object OffsetKafkaConsumerTest : Spek({
             val offsetKafkaConsumer = OffsetKafkaConsumer(mockedKafkaConsumer, emptyTopics, mockedMetrics, testDispatcher)
 
             on("call of function start") {
-                val emptyTopicPartitions = setOf(null)
+                val emptyTopicPartitions = emptySet<TopicPartition>()
                 whenever(mockedKafkaConsumer.assignment()).thenReturn(emptyTopicPartitions)
                 whenever(mockedKafkaConsumer.endOffsets(emptyTopicPartitions))
                         .thenReturn(emptyMap())
diff --git a/sources/hv-collector-kafka-consumer/src/test/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/MicrometerMetricsTest.kt b/sources/hv-collector-kafka-consumer/src/test/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/MicrometerMetricsTest.kt
index 93a39ae8..cfe67df2 100644
--- a/sources/hv-collector-kafka-consumer/src/test/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/MicrometerMetricsTest.kt
+++ b/sources/hv-collector-kafka-consumer/src/test/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/MicrometerMetricsTest.kt
@@ -69,17 +69,34 @@ object MicrometerMetricsTest : Spek({
     }
 
     describe("Gauges") {
-        val gaugeName = "$PREFIX.consumer.offset.topic"
+        val gaugeName1 = "$PREFIX.offset.partition.sample_topic-0"
+        val gaugeName2 = "$PREFIX.offset.partition.sample_topic-1"
+        val offset1 = 966L
+        val offset2 = 967L
+        val topicPartition1 = TopicPartition("sample_topic", 0)
+        val topicPartition2 = TopicPartition("sample_topic", 1)
 
         on("notifyOffsetChanged") {
-            val offset = 966L
-            val topicPartition = TopicPartition("sample_topic", 1)
+            it("should update $gaugeName1") {
+                cut.notifyOffsetChanged(offset1, topicPartition1)
 
-            it("should update $gaugeName") {
-                cut.notifyOffsetChanged(offset, topicPartition)
+                registry.verifyGauge(gaugeName1) {
+                    assertThat(it.value()).isCloseTo(offset1.toDouble(), doublePrecision)
+                }
+            }
+        }
+
+        on("two partition update") {
+            it("should update $gaugeName1") {
+                cut.notifyOffsetChanged(offset1, topicPartition1)
+                cut.notifyOffsetChanged(offset2, topicPartition2)
+
+                registry.verifyGauge(gaugeName1) {
+                    assertThat(it.value()).isCloseTo(offset1.toDouble(), doublePrecision)
+                }
 
-                registry.verifyGauge(gaugeName) {
-                    assertThat(it.value()).isCloseTo(offset.toDouble(), doublePrecision)
+                registry.verifyGauge(gaugeName2) {
+                    assertThat(it.value()).isCloseTo(offset2.toDouble(), doublePrecision)
                 }
             }
         }