From f3b6f88fe48da4808f2521cb0769cc77caf4f3fb Mon Sep 17 00:00:00 2001
From: Joanna Jeremicz <joanna.jeremicz@nokia.com>
Date: Tue, 28 Apr 2020 13:25:46 +0200
Subject: Make offset consumer more generic

Offsets are now correctly read despite of partitions number on topic.

Issue-ID: DCAEGEN2-1576
Change-Id: Id89bebbcf1b6181a2f9fd5a4a4600329d046c5e5
Signed-off-by: Joanna Jeremicz <joanna.jeremicz@nokia.com>
---
 .../collectors/veshv/kafkaconsumer/impl/OffsetKafkaConsumer.kt     | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

(limited to 'sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae')

diff --git a/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/impl/OffsetKafkaConsumer.kt b/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/impl/OffsetKafkaConsumer.kt
index 52bcf1e4..18999153 100644
--- a/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/impl/OffsetKafkaConsumer.kt
+++ b/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/impl/OffsetKafkaConsumer.kt
@@ -41,9 +41,10 @@ internal class OffsetKafkaConsumer(private val kafkaConsumer: KafkaConsumer<Byte
 
     override suspend fun start(updateInterval: Long, pollTimeout: Duration): Job =
             GlobalScope.launch(dispatcher) {
-                val topicPartitions = topics.flatMap {
-                    listOf(TopicPartition(it, 0), TopicPartition(it, 1), TopicPartition(it, 2))
-                }
+
+                val topicPartitions = topics.flatMap(kafkaConsumer::partitionsFor)
+                        .map { TopicPartition(it.topic(), it.partition()) }
+
                 kafkaConsumer.assign(topicPartitions)
 
                 while (isActive) {
-- 
cgit 1.2.3-korg