Add Message Header support for tracing.

Issue-ID: CCSDK-1046
Signed-off-by: Brinda Santh <bs2796@att.com>
Change-Id: If6a416fe157a1a8ed4d93878142b3c959d49a035

2 files changed, 14 insertions, 11 deletions

diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageConsumerServiceTest.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageConsumerServiceTest.kt
index f4e85a94b..86c2ec5ef 100644
--- a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageConsumerServiceTest.kt
+++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageConsumerServiceTest.kt
@@ -87,14 +87,14 @@ open class BlueprintMessageConsumerServiceTest {
                 partitionsEndMap[partition] = records
                 topicsCollection.add(partition.topic())
             }
-            val mockKafkaConsumer = MockConsumer<String, String>(OffsetResetStrategy.EARLIEST)
+            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, String>(topic, 1, i.toLong(), "key_$i",
-                        "I am message $i")
+                val record = ConsumerRecord<String, ByteArray>(topic, 1, i.toLong(), "key_$i",
+                        "I am message $i".toByteArray())
                 mockKafkaConsumer.addRecord(record)
             }
 
@@ -131,7 +131,10 @@ open class BlueprintMessageConsumerServiceTest {
             launch {
                 repeat(5) {
                     delay(100)
-                    blueprintMessageProducerService.sendMessage("this is my message($it)")
+                    val headers: MutableMap<String, String> = hashMapOf()
+                    headers["id"] = it.toString()
+                    blueprintMessageProducerService.sendMessageNB(message = "this is my message($it)",
+                            headers = headers)
                 }
             }
             delay(5000)
diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageProducerServiceTest.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageProducerServiceTest.kt
index 31bcc1517..f23624f7a 100644
--- a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageProducerServiceTest.kt
+++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/BlueprintMessageProducerServiceTest.kt
@@ -20,18 +20,18 @@ import io.mockk.every
 import io.mockk.mockk
 import io.mockk.spyk
 import kotlinx.coroutines.runBlocking
+import org.apache.kafka.clients.producer.KafkaProducer
+import org.apache.kafka.clients.producer.RecordMetadata
 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.springframework.beans.factory.annotation.Autowired
-import org.springframework.kafka.core.KafkaTemplate
-import org.springframework.kafka.support.SendResult
 import org.springframework.test.annotation.DirtiesContext
 import org.springframework.test.context.ContextConfiguration
 import org.springframework.test.context.TestPropertySource
 import org.springframework.test.context.junit4.SpringRunner
-import org.springframework.util.concurrent.SettableListenableFuture
+import java.util.concurrent.Future
 import kotlin.test.Test
 import kotlin.test.assertTrue
 
@@ -57,12 +57,12 @@ open class BlueprintMessageProducerServiceTest {
             val blueprintMessageProducerService = bluePrintMessageLibPropertyService
                     .blueprintMessageProducerService("sample") as KafkaBasicAuthMessageProducerService
 
-            val mockKafkaTemplate = mockk<KafkaTemplate<String, Any>>()
+            val mockKafkaTemplate = mockk<KafkaProducer<String, ByteArray>>()
 
-            val future = SettableListenableFuture<SendResult<String, Any>>()
-            //future.setException(BluePrintException("failed sending"))
+            val responseMock = mockk<Future<RecordMetadata>>()
+            every { responseMock.get() } returns mockk()
 
-            every { mockKafkaTemplate.send(any(), any()) } returns future
+            every { mockKafkaTemplate.send(any(), any()) } returns responseMock
 
             val spyBluePrintMessageProducerService = spyk(blueprintMessageProducerService, recordPrivateCalls = true)