diff options
Diffstat (limited to 'sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/impl/OffsetKafkaConsumer.kt')
-rw-r--r-- | sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/impl/OffsetKafkaConsumer.kt | 14 |
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) } } |