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

   1 /*
   2  *  Copyright © 2019 IBM.
   3  *  Modifications Copyright © 2018-2019 AT&T 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.CommonClientConfigs
  25 import org.apache.kafka.clients.consumer.Consumer
  26 import org.apache.kafka.clients.consumer.ConsumerConfig
  27 import org.apache.kafka.clients.consumer.KafkaConsumer
  28 import org.apache.kafka.common.serialization.ByteArrayDeserializer
  29 import org.apache.kafka.common.serialization.StringDeserializer
  30 import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaBasicAuthMessageConsumerProperties
  31 import org.onap.ccsdk.cds.controllerblueprints.core.logger
  32 import java.nio.charset.Charset
  33 import java.time.Duration
  34 import kotlin.concurrent.thread
  35
  36 open class KafkaBasicAuthMessageConsumerService(
  37     private val messageConsumerProperties: KafkaBasicAuthMessageConsumerProperties
  38 ) :
  39     BlueprintMessageConsumerService {
  40
  41     val log = logger(KafkaBasicAuthMessageConsumerService::class)
  42     val channel = Channel<String>()
  43     var kafkaConsumer: Consumer<String, ByteArray>? = null
  44
  45     @Volatile
  46     var keepGoing = true
  47
  48     fun kafkaConsumer(additionalConfig: Map<String, Any>? = null): Consumer<String, ByteArray> {
  49         val configProperties = hashMapOf<String, Any>()
  50         configProperties[CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG] = messageConsumerProperties.bootstrapServers
  51         configProperties[ConsumerConfig.GROUP_ID_CONFIG] = messageConsumerProperties.groupId
  52         configProperties[ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG] = messageConsumerProperties.autoCommit
  53         /**
  54          * earliest: automatically reset the offset to the earliest offset
  55          * latest: automatically reset the offset to the latest offset
  56          */
  57         configProperties[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = messageConsumerProperties.autoOffsetReset
  58         configProperties[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = StringDeserializer::class.java
  59         configProperties[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = ByteArrayDeserializer::class.java
  60         configProperties[ConsumerConfig.CLIENT_ID_CONFIG] = messageConsumerProperties.clientId
  61
  62         /** To handle Back pressure, Get only configured record for processing */
  63         if (messageConsumerProperties.pollRecords > 0) {
  64             configProperties[ConsumerConfig.MAX_POLL_RECORDS_CONFIG] = messageConsumerProperties.pollRecords
  65         }
  66         // TODO("Security Implementation based on type")
  67         /** add or override already set properties */
  68         additionalConfig?.let { configProperties.putAll(it) }
  69         /** Create Kafka consumer */
  70         return KafkaConsumer(configProperties)
  71     }
  72
  73     override suspend fun subscribe(additionalConfig: Map<String, Any>?): Channel<String> {
  74         /** get to topic names */
  75         val consumerTopic = messageConsumerProperties.topic?.split(",")?.map { it.trim() }
  76         check(!consumerTopic.isNullOrEmpty()) { "couldn't get topic information" }
  77         return subscribe(consumerTopic, additionalConfig)
  78     }
  79
  80     override suspend fun subscribe(topics: List<String>, additionalConfig: Map<String, Any>?): Channel<String> {
  81         /** Create Kafka consumer */
  82         kafkaConsumer = kafkaConsumer(additionalConfig)
  83
  84         checkNotNull(kafkaConsumer) {
  85             "failed to create kafka consumer for " +
  86                     "server(${messageConsumerProperties.bootstrapServers})'s " +
  87                     "topics(${messageConsumerProperties.bootstrapServers})"
  88         }
  89
  90         kafkaConsumer!!.subscribe(topics)
  91         log.info("Successfully consumed topic($topics)")
  92
  93         thread(start = true, name = "KafkaConsumer-${messageConsumerProperties.clientId}") {
  94             keepGoing = true
  95             kafkaConsumer!!.use { kc ->
  96                 while (keepGoing) {
  97                     val consumerRecords = kc.poll(Duration.ofMillis(messageConsumerProperties.pollMillSec))
  98                     log.trace("Consumed Records : ${consumerRecords.count()}")
  99                     runBlocking {
 100                         consumerRecords?.forEach { consumerRecord ->
 101                             /** execute the command block */
 102                             consumerRecord.value()?.let {
 103                                 launch {
 104                                     if (!channel.isClosedForSend) {
 105                                         channel.send(String(it, Charset.defaultCharset()))
 106                                     } else {
 107                                         log.error("Channel is closed to receive message")
 108                                     }
 109                                 }
 110                             }
 111                         }
 112                     }
 113                 }
 114                 log.info("message listener shutting down.....")
 115             }
 116         }
 117         return channel
 118     }
 119
 120     override suspend fun consume(additionalConfig: Map<String, Any>?, consumerFunction: ConsumerFunction) {
 121         /** get to topic names */
 122         val consumerTopic = messageConsumerProperties.topic?.split(",")?.map { it.trim() }
 123         check(!consumerTopic.isNullOrEmpty()) { "couldn't get topic information" }
 124         return consume(topics = consumerTopic, additionalConfig = additionalConfig, consumerFunction = consumerFunction)
 125     }
 126
 127     override suspend fun consume(
 128         topics: List<String>,
 129         additionalConfig: Map<String, Any>?,
 130         consumerFunction: ConsumerFunction
 131     ) {
 132
 133         val kafkaConsumerFunction = consumerFunction as KafkaConsumerRecordsFunction
 134
 135         /** Create Kafka consumer */
 136         kafkaConsumer = kafkaConsumer(additionalConfig)
 137
 138         checkNotNull(kafkaConsumer) {
 139             "failed to create kafka consumer for " +
 140                     "server(${messageConsumerProperties.bootstrapServers})'s " +
 141                     "topics(${messageConsumerProperties.bootstrapServers})"
 142         }
 143
 144         kafkaConsumer!!.subscribe(topics)
 145         log.info("Successfully consumed topic($topics)")
 146
 147         thread(start = true, name = "KafkaConsumer-${messageConsumerProperties.clientId}") {
 148             keepGoing = true
 149             kafkaConsumer!!.use { kc ->
 150                 while (keepGoing) {
 151                     val consumerRecords = kc.poll(Duration.ofMillis(messageConsumerProperties.pollMillSec))
 152                     log.trace("Consumed Records : ${consumerRecords.count()}")
 153                     runBlocking {
 154                         /** Execute dynamic consumer Block substitution */
 155                         kafkaConsumerFunction.invoke(messageConsumerProperties, kc, consumerRecords)
 156                     }
 157                 }
 158                 log.info("message listener shutting down.....")
 159             }
 160         }
 161     }
 162
 163     override suspend fun shutDown() {
 164         /** stop the polling loop */
 165         keepGoing = false
 166         /** Close the Channel */
 167         channel.cancel()
 168         /** TO shutdown gracefully, need to wait for the maximum poll time */
 169         delay(messageConsumerProperties.pollMillSec)
 170     }
 171 }