ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaStreamsConsumerServiceTest.kt

   1 /*
   2  * Copyright © 2018-2019 AT&T Intellectual Property.
   3  * Modifications Copyright © 2021 Bell Canada.
   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 io.micrometer.core.instrument.MeterRegistry
  21 import kotlinx.coroutines.delay
  22 import kotlinx.coroutines.launch
  23 import kotlinx.coroutines.runBlocking
  24 import org.apache.kafka.common.serialization.Serdes
  25 import org.apache.kafka.streams.Topology
  26 import org.apache.kafka.streams.processor.Processor
  27 import org.apache.kafka.streams.processor.ProcessorSupplier
  28 import org.apache.kafka.streams.state.Stores
  29 import org.junit.Test
  30 import org.junit.runner.RunWith
  31 import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
  32 import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
  33 import org.onap.ccsdk.cds.blueprintsprocessor.message.BluePrintMessageLibConfiguration
  34 import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaStreamsBasicAuthConsumerProperties
  35 import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageConsumerProperties
  36 import org.springframework.beans.factory.annotation.Autowired
  37 import org.springframework.boot.test.mock.mockito.MockBean
  38 import org.springframework.test.annotation.DirtiesContext
  39 import org.springframework.test.context.ContextConfiguration
  40 import org.springframework.test.context.TestPropertySource
  41 import org.springframework.test.context.junit4.SpringRunner
  42 import kotlin.test.assertNotNull
  43
  44 @RunWith(SpringRunner::class)
  45 @DirtiesContext
  46 @ContextConfiguration(
  47     classes = [
  48         BluePrintMessageLibConfiguration::class,
  49         BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class
  50     ]
  51 )
  52 @TestPropertySource(
  53     properties =
  54         [
  55             "blueprintsprocessor.messageproducer.sample.type=kafka-scram-ssl-auth",
  56             "blueprintsprocessor.messageproducer.sample.bootstrapServers=127.0.0.1:9092",
  57             "blueprintsprocessor.messageproducer.sample.topic=default-stream-topic",
  58             "blueprintsprocessor.messageproducer.sample.clientId=default-client-id",
  59             "blueprintsprocessor.messageproducer.sample.truststore=/path/to/truststore.jks",
  60             "blueprintsprocessor.messageproducer.sample.truststorePassword=secretpassword",
  61             "blueprintsprocessor.messageproducer.sample.scramUsername=sample-user",
  62             "blueprintsprocessor.messageproducer.sample.scramPassword=secretpassword",
  63
  64             "blueprintsprocessor.messageconsumer.stream-consumer.type=kafka-streams-scram-ssl-auth",
  65             "blueprintsprocessor.messageconsumer.stream-consumer.bootstrapServers=127.0.0.1:9092",
  66             "blueprintsprocessor.messageconsumer.stream-consumer.applicationId=test-streams-application",
  67             "blueprintsprocessor.messageconsumer.stream-consumer.topic=default-stream-topic",
  68             "blueprintsprocessor.messageproducer.stream-consumer.truststore=/path/to/truststore.jks",
  69             "blueprintsprocessor.messageproducer.stream-consumer.truststorePassword=secretpassword",
  70             "blueprintsprocessor.messageproducer.stream-consumer.scramUsername=sample-user",
  71             "blueprintsprocessor.messageproducer.stream-consumer.scramPassword=secretpassword"
  72
  73         ]
  74 )
  75 class KafkaStreamsConsumerServiceTest {
  76
  77     @Autowired
  78     lateinit var bluePrintMessageLibPropertyService: BluePrintMessageLibPropertyService
  79
  80     @MockBean
  81     lateinit var meterRegistry: MeterRegistry
  82
  83     @Test
  84     fun testProperties() {
  85         val blueprintMessageConsumerService = bluePrintMessageLibPropertyService.blueprintMessageConsumerService("stream-consumer")
  86         assertNotNull(blueprintMessageConsumerService, "failed to get blueprintMessageProducerService")
  87     }
  88
  89     /** Integration Kafka Testing, Enable and use this test case only for local desktop testing with real kafka broker */
  90     // @Test
  91     fun testKafkaStreamingMessageConsumer() {
  92         runBlocking {
  93             val streamingConsumerService = bluePrintMessageLibPropertyService.blueprintMessageConsumerService("stream-consumer")
  94
  95             // Dynamic Consumer Function to create Topology
  96             val consumerFunction = object : KafkaStreamConsumerFunction {
  97                 override suspend fun createTopology(
  98                     messageConsumerProperties: MessageConsumerProperties,
  99                     additionalConfig: Map<String, Any>?
 100                 ): Topology {
 101                     val topology = Topology()
 102                     val kafkaStreamsBasicAuthConsumerProperties = messageConsumerProperties
 103                         as KafkaStreamsBasicAuthConsumerProperties
 104
 105                     val topics = kafkaStreamsBasicAuthConsumerProperties.topic.split(",")
 106                     topology.addSource("Source", *topics.toTypedArray())
 107                     // Processor Supplier
 108                     val firstProcessorSupplier = object : ProcessorSupplier<ByteArray, ByteArray> {
 109                         override fun get(): Processor<ByteArray, ByteArray> {
 110                             return FirstProcessor()
 111                         }
 112                     }
 113                     val changelogConfig: MutableMap<String, String> = hashMapOf()
 114                     changelogConfig.put("min.insync.replicas", "1")
 115
 116                     // Store Buolder
 117                     val countStoreSupplier = Stores.keyValueStoreBuilder(
 118                         Stores.persistentKeyValueStore("PriorityMessageState"),
 119                         Serdes.String(),
 120                         PriorityMessageSerde()
 121                     )
 122                         .withLoggingEnabled(changelogConfig)
 123
 124                     topology.addProcessor("FirstProcessor", firstProcessorSupplier, "Source")
 125                     topology.addStateStore(countStoreSupplier, "FirstProcessor")
 126                     topology.addSink(
 127                         "SINK", "default-stream-topic-out", Serdes.String().serializer(),
 128                         PriorityMessageSerde().serializer(), "FirstProcessor"
 129                     )
 130                     return topology
 131                 }
 132             }
 133
 134             /** Send message with every 1 sec */
 135             val blueprintMessageProducerService = bluePrintMessageLibPropertyService
 136                 .blueprintMessageProducerService("sample") as KafkaMessageProducerService
 137             launch {
 138                 repeat(5) {
 139                     delay(1000)
 140                     val headers: MutableMap<String, String> = hashMapOf()
 141                     headers["id"] = it.toString()
 142                     blueprintMessageProducerService.sendMessageNB(
 143                         key = "mykey",
 144                         message = "this is my message($it)",
 145                         headers = headers
 146                     )
 147                 }
 148             }
 149             streamingConsumerService.consume(null, consumerFunction)
 150             delay(10000)
 151             streamingConsumerService.shutDown()
 152         }
 153     }
 154 }