Make use of Kafka Key for Audit service and Kafka listener

* When message is sent by audit service, key will be the  CBA name
* When sent by kafka listener (self-service api), key is the same as the request message key consumed. If not specified, a random UUID
* MessageProducer interface refactoring :
  * add 'key' parameter to specify a key
  * add default value null to paramater 'headers' to remove some unnecessary method

Issue-ID: CCSDK-2628
Signed-off-by: Julien Fontaine <julien.fontaine@bell.ca>
Change-Id: I68580151184c87104c07037f379276dd8c8c71c7

3 files changed, 10 insertions, 2 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 fdf6e48e7..77bdbe408 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
@@ -51,6 +51,7 @@ 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 java.nio.charset.Charset
 import kotlin.test.assertEquals
 import kotlin.test.assertNotNull
 import kotlin.test.assertTrue
@@ -133,9 +134,14 @@ open class BlueprintMessageConsumerServiceTest {
 
             every { spyBlueprintMessageConsumerService.kafkaConsumer(any()) } returns mockKafkaConsumer
             val channel = spyBlueprintMessageConsumerService.subscribe(null)
+            var i = 0
             launch {
                 channel.consumeEach {
-                    assertTrue(it.startsWith("I am message"), "failed to get the actual message")
+                    ++i
+                    val key = it.key()
+                    val value = String(it.value(), Charset.defaultCharset())
+                    assertTrue(value.startsWith("I am message"), "failed to get the actual message")
+                    assertEquals("key_$i", key)
                 }
             }
             delay(10)
@@ -268,6 +274,7 @@ open class BlueprintMessageConsumerServiceTest {
                     val headers: MutableMap<String, String> = hashMapOf()
                     headers["id"] = it.toString()
                     blueprintMessageProducerService.sendMessageNB(
+                        key = "mykey",
                         message = "this is my message($it)",
                         headers = headers
                     )
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 537dab1ba..881f0b422 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
@@ -88,7 +88,7 @@ open class BlueprintMessageProducerServiceTest {
 
             every { spyBluePrintMessageProducerService.messageTemplate(any()) } returns mockKafkaTemplate
 
-            val response = spyBluePrintMessageProducerService.sendMessage("Testing message")
+            val response = spyBluePrintMessageProducerService.sendMessage("mykey", "Testing message")
             assertTrue(response, "failed to get command response")
         }
     }
diff --git a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaStreamsConsumerServiceTest.kt b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaStreamsConsumerServiceTest.kt
index c30ab9b02..44990ae7f 100644
--- a/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaStreamsConsumerServiceTest.kt
+++ b/ms/blueprintsprocessor/modules/commons/message-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/message/service/KafkaStreamsConsumerServiceTest.kt
@@ -132,6 +132,7 @@ class KafkaStreamsConsumerServiceTest {
                     val headers: MutableMap<String, String> = hashMapOf()
                     headers["id"] = it.toString()
                     blueprintMessageProducerService.sendMessageNB(
+                        key = "mykey",
                         message = "this is my message($it)",
                         headers = headers
                     )