/*
* Copyright © 2019 IBM.
+ * Modifications Copyright © 2018-2019 AT&T Intellectual Property.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
open class KafkaMessageConsumerProperties : MessageConsumerProperties() {
lateinit var bootstrapServers: String
lateinit var groupId: String
- var clientId: String? = null
+ lateinit var clientId: String
var topic: String? = null
var autoCommit: Boolean = true
var autoOffsetReset: String = "latest"
/*
* Copyright © 2019 IBM.
+ * Modifications Copyright © 2018-2019 AT&T Intellectual Property.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
package org.onap.ccsdk.cds.blueprintsprocessor.message.service
import kotlinx.coroutines.channels.Channel
+import org.apache.kafka.clients.consumer.Consumer
+import org.apache.kafka.clients.consumer.ConsumerRecords
+import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageConsumerProperties
+import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
+
+/** Consumer Function Interfaces */
+interface ConsumerFunction
interface BlueprintMessageConsumerService {
+ suspend fun subscribe(): Channel<String> {
+ return subscribe(null)
+ }
+
/** Subscribe to the Kafka channel with [additionalConfig] */
suspend fun subscribe(additionalConfig: Map<String, Any>?): Channel<String>
/** Subscribe to the Kafka channel with [additionalConfig] for dynamic [topics]*/
suspend fun subscribe(topics: List<String>, additionalConfig: Map<String, Any>? = null): Channel<String>
+ /** Consume and execute dynamic function [consumerFunction] */
+ suspend fun consume(consumerFunction: ConsumerFunction) {
+ consume(null, consumerFunction)
+ }
+
+ /** Consume with [additionalConfig], so that we can execute dynamic function [consumerFunction] */
+ suspend fun consume(additionalConfig: Map<String, Any>?, consumerFunction: ConsumerFunction) {
+ throw BluePrintProcessorException("Not Implemented")
+ }
+
+ /** Consume the [topics] with [additionalConfig], so that we can execute dynamic function [consumerFunction] */
+ suspend fun consume(topics: List<String>, additionalConfig: Map<String, Any>?,
+ consumerFunction: ConsumerFunction) {
+ throw BluePrintProcessorException("Not Implemented")
+ }
+
/** close the channel, consumer and other resources */
suspend fun shutDown()
-
+}
+/** Consumer dynamic implementation interface */
+interface KafkaConsumerRecordsFunction : ConsumerFunction {
+ suspend fun invoke(messageConsumerProperties: MessageConsumerProperties, consumer: Consumer<*, *>,
+ consumerRecords: ConsumerRecords<*, *>)
}
\ No newline at end of file
: BlueprintMessageConsumerService {
val log = logger(KafkaBasicAuthMessageConsumerService::class)
-
val channel = Channel<String>()
var kafkaConsumer: Consumer<String, ByteArray>? = null
configProperties[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = messageConsumerProperties.autoOffsetReset
configProperties[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = StringDeserializer::class.java
configProperties[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = ByteArrayDeserializer::class.java
- if (messageConsumerProperties.clientId != null) {
- configProperties[ConsumerConfig.CLIENT_ID_CONFIG] = messageConsumerProperties.clientId!!
- }
+ configProperties[ConsumerConfig.CLIENT_ID_CONFIG] = messageConsumerProperties.clientId
+
/** To handle Back pressure, Get only configured record for processing */
if (messageConsumerProperties.pollRecords > 0) {
configProperties[ConsumerConfig.MAX_POLL_RECORDS_CONFIG] = messageConsumerProperties.pollRecords
}
- override suspend fun subscribe(consumerTopic: List<String>, additionalConfig: Map<String, Any>?): Channel<String> {
+ override suspend fun subscribe(topics: List<String>, additionalConfig: Map<String, Any>?): Channel<String> {
/** Create Kafka consumer */
kafkaConsumer = kafkaConsumer(additionalConfig)
"topics(${messageConsumerProperties.bootstrapServers})"
}
- kafkaConsumer!!.subscribe(consumerTopic)
- log.info("Successfully consumed topic($consumerTopic)")
+ kafkaConsumer!!.subscribe(topics)
+ log.info("Successfully consumed topic($topics)")
- thread(start = true, name = "KafkaConsumer") {
+ thread(start = true, name = "KafkaConsumer-${messageConsumerProperties.clientId}") {
keepGoing = true
kafkaConsumer!!.use { kc ->
while (keepGoing) {
val consumerRecords = kc.poll(Duration.ofMillis(messageConsumerProperties.pollMillSec))
- log.info("Consumed Records : ${consumerRecords.count()}")
+ log.trace("Consumed Records : ${consumerRecords.count()}")
runBlocking {
consumerRecords?.forEach { consumerRecord ->
/** execute the command block */
return channel
}
+ override suspend fun consume(additionalConfig: Map<String, Any>?, consumerFunction: ConsumerFunction) {
+ /** get to topic names */
+ val consumerTopic = messageConsumerProperties.topic?.split(",")?.map { it.trim() }
+ check(!consumerTopic.isNullOrEmpty()) { "couldn't get topic information" }
+ return consume(topics = consumerTopic, additionalConfig = additionalConfig, consumerFunction = consumerFunction)
+ }
+
+ override suspend fun consume(topics: List<String>, additionalConfig: Map<String, Any>?,
+ consumerFunction: ConsumerFunction) {
+
+ val kafkaConsumerFunction = consumerFunction as KafkaConsumerRecordsFunction
+
+ /** Create Kafka consumer */
+ kafkaConsumer = kafkaConsumer(additionalConfig)
+
+ checkNotNull(kafkaConsumer) {
+ "failed to create kafka consumer for " +
+ "server(${messageConsumerProperties.bootstrapServers})'s " +
+ "topics(${messageConsumerProperties.bootstrapServers})"
+ }
+
+ kafkaConsumer!!.subscribe(topics)
+ log.info("Successfully consumed topic($topics)")
+
+ thread(start = true, name = "KafkaConsumer-${messageConsumerProperties.clientId}") {
+ keepGoing = true
+ kafkaConsumer!!.use { kc ->
+ while (keepGoing) {
+ val consumerRecords = kc.poll(Duration.ofMillis(messageConsumerProperties.pollMillSec))
+ log.trace("Consumed Records : ${consumerRecords.count()}")
+ runBlocking {
+ /** Execute dynamic consumer Block substitution */
+ kafkaConsumerFunction.invoke(messageConsumerProperties, kc, consumerRecords)
+ }
+ }
+ log.info("message listener shutting down.....")
+ }
+ }
+ }
+
override suspend fun shutDown() {
/** stop the polling loop */
keepGoing = false
/*
* Copyright © 2019 IBM.
+ * Modifications Copyright © 2018-2019 AT&T Intellectual Property.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
-import org.apache.kafka.clients.consumer.ConsumerRecord
-import org.apache.kafka.clients.consumer.MockConsumer
-import org.apache.kafka.clients.consumer.OffsetResetStrategy
+import org.apache.kafka.clients.consumer.*
import org.apache.kafka.common.TopicPartition
import org.junit.Test
import org.junit.runner.RunWith
import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintProperties
import org.onap.ccsdk.cds.blueprintsprocessor.core.BlueprintPropertyConfiguration
import org.onap.ccsdk.cds.blueprintsprocessor.message.BluePrintMessageLibConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageConsumerProperties
import org.onap.ccsdk.cds.controllerblueprints.core.logger
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.test.annotation.DirtiesContext
}
}
+ @Test
+ fun testKafkaBasicAuthConsumerWithDynamicFunction() {
+ runBlocking {
+ val blueprintMessageConsumerService = bluePrintMessageLibPropertyService
+ .blueprintMessageConsumerService("sample") as KafkaBasicAuthMessageConsumerService
+ assertNotNull(blueprintMessageConsumerService, "failed to get blueprintMessageConsumerService")
+
+ val spyBlueprintMessageConsumerService = spyk(blueprintMessageConsumerService, recordPrivateCalls = true)
+
+ val topic = "default-topic"
+ val partitions: MutableList<TopicPartition> = arrayListOf()
+ val topicsCollection: MutableList<String> = arrayListOf()
+ partitions.add(TopicPartition(topic, 1))
+ val partitionsBeginningMap: MutableMap<TopicPartition, Long> = mutableMapOf()
+ val partitionsEndMap: MutableMap<TopicPartition, Long> = mutableMapOf()
+
+ val records: Long = 10
+ partitions.forEach { partition ->
+ partitionsBeginningMap[partition] = 0L
+ partitionsEndMap[partition] = records
+ topicsCollection.add(partition.topic())
+ }
+ val mockKafkaConsumer = MockConsumer<String, ByteArray>(OffsetResetStrategy.EARLIEST)
+ mockKafkaConsumer.subscribe(topicsCollection)
+ mockKafkaConsumer.rebalance(partitions)
+ mockKafkaConsumer.updateBeginningOffsets(partitionsBeginningMap)
+ mockKafkaConsumer.updateEndOffsets(partitionsEndMap)
+ for (i in 1..10) {
+ val record = ConsumerRecord<String, ByteArray>(topic, 1, i.toLong(), "key_$i",
+ "I am message $i".toByteArray())
+ mockKafkaConsumer.addRecord(record)
+ }
+
+ every { spyBlueprintMessageConsumerService.kafkaConsumer(any()) } returns mockKafkaConsumer
+ /** Test Consumer Function implementation */
+ val consumerFunction = object : KafkaConsumerRecordsFunction {
+ override suspend fun invoke(messageConsumerProperties: MessageConsumerProperties,
+ consumer: Consumer<*, *>, consumerRecords: ConsumerRecords<*, *>) {
+ val count = consumerRecords.count()
+ log.trace("Received Message count($count)")
+ }
+ }
+ spyBlueprintMessageConsumerService.consume(consumerFunction)
+ delay(10)
+ spyBlueprintMessageConsumerService.shutDown()
+ }
+ }
+
/** Integration Kafka Testing, Enable and use this test case only for local desktop testing with real kafka broker */
//@Test
fun testKafkaIntegration() {
<!-- encoders are assigned the type
ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->
<encoder>
- <pattern>${testing}</pattern>
+ <pattern>${localPattern}</pattern>
</encoder>
</appender>
Code Review / ccsdk / cds.git / commitdiff