From 3e4f108d66fb31e5ea713fef1cf7bc99c0510c21 Mon Sep 17 00:00:00 2001 From: kjaniak Date: Wed, 26 Jun 2019 15:57:29 +0200 Subject: Reorganize kafka module Change-Id: I2eb7a8a6e92c9d89586b877f4cae438497b62ae2 Issue-ID: DCAEGEN2-1635 Signed-off-by: kjaniak --- .../veshv/kafkaconsumer/impl/KafkaSource.kt | 48 ++++++++++++++++++++++ .../veshv/kafkaconsumer/metrics/Metrics.kt | 6 +-- .../kafkaconsumer/metrics/MicrometerMetrics.kt | 3 +- .../veshv/kafkaconsumer/state/OffsetConsumer.kt | 19 +++++---- 4 files changed, 63 insertions(+), 13 deletions(-) create mode 100644 sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/impl/KafkaSource.kt (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/KafkaSource.kt b/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/impl/KafkaSource.kt new file mode 100644 index 00000000..dd24345d --- /dev/null +++ b/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/impl/KafkaSource.kt @@ -0,0 +1,48 @@ +/* + * ============LICENSE_START======================================================= + * dcaegen2-collectors-veshv + * ================================================================================ + * Copyright (C) 2019 NOKIA + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END========================================================= + */ +package org.onap.dcae.collectors.veshv.kafkaconsumer.impl + +import kotlinx.coroutines.CoroutineDispatcher +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.GlobalScope +import kotlinx.coroutines.Job +import kotlinx.coroutines.delay +import kotlinx.coroutines.isActive +import kotlinx.coroutines.launch +import org.apache.kafka.clients.consumer.KafkaConsumer +import org.onap.dcae.collectors.veshv.kafkaconsumer.state.OffsetConsumer + +internal class KafkaSource(private val kafkaConsumer: KafkaConsumer, + private val topics: Set, + private val dispatcher: CoroutineDispatcher = Dispatchers.IO) { + suspend fun start(offsetConsumer: OffsetConsumer, updateInterval: Long = 500L): Job = + GlobalScope.launch(dispatcher) { + kafkaConsumer.subscribe(topics) + val topicPartitions = kafkaConsumer.assignment() + while (isActive) { + kafkaConsumer.endOffsets(topicPartitions) + .forEach { (topicPartition, offset) -> + offsetConsumer.update(topicPartition, offset) + } + kafkaConsumer.commitSync() + delay(updateInterval) + } + } +} diff --git a/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/Metrics.kt b/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/Metrics.kt index 2fabf30e..e576a88f 100644 --- a/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/Metrics.kt +++ b/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/Metrics.kt @@ -19,7 +19,7 @@ */ package org.onap.dcae.collectors.veshv.kafkaconsumer.metrics -internal interface Metrics { - fun notifyOffsetChanged(offset: Long) +interface Metrics { + fun notifyOffsetChanged(offset: Long, topic: String, partition: Int = 0) fun notifyMessageTravelTime(messageSentTimeMicros: Long) -} \ No newline at end of file +} diff --git a/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/MicrometerMetrics.kt b/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/MicrometerMetrics.kt index 748e43fc..da1225e9 100644 --- a/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/MicrometerMetrics.kt +++ b/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/metrics/MicrometerMetrics.kt @@ -41,7 +41,8 @@ internal class MicrometerMetrics constructor( registry.scrape() } - override fun notifyOffsetChanged(offset: Long) { + override fun notifyOffsetChanged(offset: Long, topic: String, partition: Int) { + // TODO use topic and partition currentOffset.lazySet(offset) } diff --git a/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/state/OffsetConsumer.kt b/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/state/OffsetConsumer.kt index 2c6707f9..1481a224 100644 --- a/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/state/OffsetConsumer.kt +++ b/sources/hv-collector-kafka-consumer/src/main/kotlin/org/onap/dcae/collectors/veshv/kafkaconsumer/state/OffsetConsumer.kt @@ -19,23 +19,24 @@ */ package org.onap.dcae.collectors.veshv.kafkaconsumer.state -import org.apache.kafka.clients.consumer.ConsumerRecord -import org.onap.dcae.collectors.veshv.kafka.api.KafkaConsumer +import org.apache.kafka.common.TopicPartition import org.onap.dcae.collectors.veshv.kafkaconsumer.metrics.Metrics import org.onap.dcae.collectors.veshv.utils.logging.Logger -internal class OffsetConsumer(private val metrics: Metrics): KafkaConsumer { +internal class OffsetConsumer(private val metrics: Metrics) { - override fun update(record: ConsumerRecord) { - val offset = record.offset() - logger.trace { "Current consumer offset $offset" } - metrics.notifyOffsetChanged(offset) + fun update(topicPartition: TopicPartition, offset: Long) { + logger.trace { + "Current consumer offset $offset for topic ${topicPartition.topic()} " + + "on partition ${topicPartition.partition()}" + } + metrics.notifyOffsetChanged(offset, topicPartition.topic(), topicPartition.partition()) } - override fun reset() = Unit + fun reset() = Unit companion object { - private val logger = Logger(OffsetConsumer::class) + val logger = Logger(OffsetConsumer::class) } } -- cgit 1.2.3-korg