diff options
author | 2019-09-24 09:46:24 +0200 | |
---|---|---|
committer | 2019-09-26 14:34:14 +0200 | |
commit | 728ca6351b290a15cb7fec07f56e0752f7addd70 (patch) | |
tree | 62314472bf5e5180f671e8f8a45239d4bf46c674 /sources/hv-collector-kafka-consumer/src/test/kotlin/org/onap/dcae | |
parent | 6d815b760e927bace06451cf72972a505a080b6c (diff) |
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>
Diffstat (limited to 'sources/hv-collector-kafka-consumer/src/test/kotlin/org/onap/dcae')
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) } } } |