1 files changed, 9 insertions, 5 deletions
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 d105c4a7..18de6fcc 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
@@ -28,19 +28,23 @@ import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 import org.apache.kafka.clients.consumer.KafkaConsumer
 import org.apache.kafka.common.TopicPartition
+import org.onap.dcae.collectors.veshv.kafkaconsumer.api.MetricsKafkaConsumer
 import org.onap.dcae.collectors.veshv.kafkaconsumer.metrics.Metrics
 import org.onap.dcae.collectors.veshv.utils.logging.Logger
+import java.time.Duration
 
 internal class OffsetKafkaConsumer(private val kafkaConsumer: KafkaConsumer<ByteArray, ByteArray>,
                                    private val topics: Set<String>,
                                    private val metrics: Metrics,
-                                   private val dispatcher: CoroutineDispatcher = Dispatchers.IO) {
+                                   private val dispatcher: CoroutineDispatcher = Dispatchers.IO)
+    : MetricsKafkaConsumer{
 
-    suspend fun start(updateInterval: Long = 500L): Job =
+    override suspend fun start(updateInterval: Long, pollTimeout: Duration):Job =
             GlobalScope.launch(dispatcher) {
-                kafkaConsumer.subscribe(topics)
-                    val topicPartitions = kafkaConsumer.assignment()
+                kafkaConsumer.assign(topics.map { TopicPartition(it, 0) })
                     while (isActive) {
+                        val topicPartitions = kafkaConsumer.assignment()
+
                         kafkaConsumer.endOffsets(topicPartitions)
                                 .forEach { (topicPartition, offset) ->
                                     update(topicPartition, offset)
@@ -58,6 +62,6 @@ internal class OffsetKafkaConsumer(private val kafkaConsumer: KafkaConsumer<Byte
     }
 
     companion object {
-        val logger = Logger(OffsetKafkaConsumer::class)
+        private val logger = Logger(OffsetKafkaConsumer::class)
     }
 }