gerrit.onap Code Review - ccsdk/cds.git/blob

   1 /*
   2  *  Copyright © 2019 IBM.
   3  *  Modifications Copyright © 2018-2021 AT&T, Bell Canada Intellectual Property.
   4  *
   5  *  Licensed under the Apache License, Version 2.0 (the "License");
   6  *  you may not use this file except in compliance with the License.
   7  *  You may obtain a copy of the License at
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  *  Unless required by applicable law or agreed to in writing, software
  12  *  distributed under the License is distributed on an "AS IS" BASIS,
  13  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  *  See the License for the specific language governing permissions and
  15  *  limitations under the License.
  16  */
  17
  18 package org.onap.ccsdk.cds.blueprintsprocessor.message.service
  19
  20 import kotlinx.coroutines.channels.Channel
  21 import kotlinx.coroutines.delay
  22 import kotlinx.coroutines.launch
  23 import kotlinx.coroutines.runBlocking
  24 import org.apache.kafka.clients.consumer.Consumer
  25 import org.apache.kafka.clients.consumer.ConsumerRecord
  26 import org.apache.kafka.clients.consumer.KafkaConsumer
  27 import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaBasicAuthMessageConsumerProperties
  28 import org.onap.ccsdk.cds.controllerblueprints.core.logger
  29 import java.time.Duration
  30 import kotlin.concurrent.thread
  31
  32 open class KafkaMessageConsumerService(
  33     private val messageConsumerProperties: KafkaBasicAuthMessageConsumerProperties
  34 ) :
  35     BlueprintMessageConsumerService {
  36
  37     val log = logger(KafkaMessageConsumerService::class)
  38     val channel = Channel<ConsumerRecord<String, ByteArray>>()
  39     var kafkaConsumer: Consumer<String, ByteArray>? = null
  40
  41     @Volatile
  42     var keepGoing = true
  43
  44     fun kafkaConsumer(additionalConfig: Map<String, Any>? = null): Consumer<String, ByteArray> {
  45         val configProperties = messageConsumerProperties.getConfig()
  46         /** add or override already set properties */
  47         additionalConfig?.let { configProperties.putAll(it) }
  48         /** Create Kafka consumer */
  49         return KafkaConsumer(configProperties)
  50     }
  51
  52     override suspend fun subscribe(additionalConfig: Map<String, Any>?): Channel<ConsumerRecord<String, ByteArray>> {
  53         /** get to topic names */
  54         val consumerTopic = messageConsumerProperties.topic?.split(",")?.map { it.trim() }
  55         check(!consumerTopic.isNullOrEmpty()) { "couldn't get topic information" }
  56         return subscribe(consumerTopic, additionalConfig)
  57     }
  58
  59     override suspend fun subscribe(topics: List<String>, additionalConfig: Map<String, Any>?): Channel<ConsumerRecord<String, ByteArray>> {
  60         /** Create Kafka consumer */
  61         kafkaConsumer = kafkaConsumer(additionalConfig)
  62
  63         checkNotNull(kafkaConsumer) {
  64             "failed to create kafka consumer for " +
  65                 "server(${messageConsumerProperties.bootstrapServers})'s " +
  66                 "topics(${messageConsumerProperties.bootstrapServers})"
  67         }
  68
  69         kafkaConsumer!!.subscribe(topics)
  70         log.info("Successfully consumed topic($topics)")
  71
  72         thread(start = true, name = "KafkaConsumer-${messageConsumerProperties.clientId}") {
  73             keepGoing = true
  74             kafkaConsumer!!.use { kc ->
  75                 while (keepGoing) {
  76                     val consumerRecords = kc.poll(Duration.ofMillis(messageConsumerProperties.pollMillSec))
  77                     log.trace("Consumed Records : ${consumerRecords.count()}")
  78                     runBlocking {
  79                         consumerRecords?.forEach { consumerRecord ->
  80                             launch {
  81                                 /** execute the command block */
  82                                 if (!channel.isClosedForSend) {
  83                                     channel.send(consumerRecord)
  84                                     log.info(
  85                                         "Channel sent Consumer Record : topic(${consumerRecord.topic()}) " +
  86                                             "partition(${consumerRecord.partition()}) " +
  87                                             "leaderEpoch(${consumerRecord.leaderEpoch().get()}) " +
  88                                             "offset(${consumerRecord.offset()}) " +
  89                                             "key(${consumerRecord.key()})"
  90                                     )
  91                                 } else {
  92                                     log.error("Channel is closed to receive message")
  93                                 }
  94                             }
  95                         }
  96                     }
  97                 }
  98                 log.info("message listener shutting down.....")
  99             }
 100         }
 101         return channel
 102     }
 103
 104     override suspend fun consume(additionalConfig: Map<String, Any>?, consumerFunction: ConsumerFunction) {
 105         /** get to topic names */
 106         val consumerTopic = messageConsumerProperties.topic?.split(",")?.map { it.trim() }
 107         check(!consumerTopic.isNullOrEmpty()) { "couldn't get topic information" }
 108         return consume(topics = consumerTopic, additionalConfig = additionalConfig, consumerFunction = consumerFunction)
 109     }
 110
 111     override suspend fun consume(
 112         topics: List<String>,
 113         additionalConfig: Map<String, Any>?,
 114         consumerFunction: ConsumerFunction
 115     ) {
 116
 117         val kafkaConsumerFunction = consumerFunction as KafkaConsumerRecordsFunction
 118
 119         /** Create Kafka consumer */
 120         kafkaConsumer = kafkaConsumer(additionalConfig)
 121
 122         checkNotNull(kafkaConsumer) {
 123             "failed to create kafka consumer for " +
 124                 "server(${messageConsumerProperties.bootstrapServers})'s " +
 125                 "topics(${messageConsumerProperties.bootstrapServers})"
 126         }
 127
 128         kafkaConsumer!!.subscribe(topics)
 129         log.info("Successfully consumed topic($topics)")
 130
 131         thread(start = true, name = "KafkaConsumer-${messageConsumerProperties.clientId}") {
 132             keepGoing = true
 133             kafkaConsumer!!.use { kc ->
 134                 while (keepGoing) {
 135                     val consumerRecords = kc.poll(Duration.ofMillis(messageConsumerProperties.pollMillSec))
 136                     log.trace("Consumed Records : ${consumerRecords.count()}")
 137                     runBlocking {
 138                         /** Execute dynamic consumer Block substitution */
 139                         kafkaConsumerFunction.invoke(messageConsumerProperties, kc, consumerRecords)
 140                     }
 141                 }
 142                 log.info("message listener shutting down.....")
 143             }
 144         }
 145     }
 146
 147     override suspend fun shutDown() {
 148         /** stop the polling loop */
 149         keepGoing = false
 150         /** Close the Channel */
 151         channel.cancel()
 152         /** TO shutdown gracefully, need to wait for the maximum poll time */
 153         delay(messageConsumerProperties.pollMillSec)
 154     }
 155 }