Reorganize kafka module

Change-Id: I2eb7a8a6e92c9d89586b877f4cae438497b62ae2
Issue-ID: DCAEGEN2-1635
Signed-off-by: kjaniak <kornel.janiak@nokia.com>

4 files changed, 63 insertions, 13 deletions

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<ByteArray, ByteArray>,
+                           private val topics: Set<String>,
+                           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<ByteArray, ByteArray>) {
-        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)
     }
 }