Merge "Include correlationId in group lock."

2 files changed, 16 insertions, 7 deletions

diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizeExtensions.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizeExtensions.kt
index 39d081455..bef7a7b61 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizeExtensions.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizeExtensions.kt
@@ -36,14 +36,18 @@ fun AbstractComponentFunction.messagePrioritizationStateService() =
 /**
  * MessagePrioritization correlation extensions
  */
+
+/**
+ * Arrange comma separated correlation keys in ascending order.
+ */
 fun MessagePrioritization.toFormatedCorrelation(): String {
-    val ascendingKey = this.correlationId!!.split(",")
+    return this.correlationId!!.split(",")
         .map { it.trim() }.sorted().joinToString(",")
-    return ascendingKey
 }
 
+/**
+ * Used to group the correlation with respect to types.
+ */
 fun MessagePrioritization.toTypeNCorrelation(): TypeCorrelationKey {
-    val ascendingKey = this.correlationId!!.split(",")
-        .map { it.trim() }.sorted().joinToString(",")
-    return TypeCorrelationKey(this.type, ascendingKey)
+    return TypeCorrelationKey(this.type, this.toFormatedCorrelation())
 }
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageProcessorUtils.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageProcessorUtils.kt
index d1f38f4f2..49230b6e4 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageProcessorUtils.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageProcessorUtils.kt
@@ -22,6 +22,7 @@ import org.onap.ccsdk.cds.blueprintsprocessor.core.service.ClusterLock
 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.AbstractMessagePrioritizeProcessor
 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.PrioritizationConfiguration
 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.toFormatedCorrelation
 import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
 import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService
 
@@ -32,8 +33,12 @@ object MessageProcessorUtils {
         clusterService: BluePrintClusterService?,
         messagePrioritization: MessagePrioritization
     ): ClusterLock? {
-        return if (clusterService != null && clusterService.clusterJoined()) {
-            val lockName = "prioritization-${messagePrioritization.group}"
+        return if (clusterService != null && clusterService.clusterJoined() &&
+            !messagePrioritization.correlationId.isNullOrBlank()
+        ) {
+            // Get the correlation key in ascending order, even it it is misplaced
+            val correlationId = messagePrioritization.toFormatedCorrelation()
+            val lockName = "prioritization-${messagePrioritization.group}-$correlationId"
             val clusterLock = clusterService.clusterLock(lockName)
             clusterLock.lock()
             if (!clusterLock.isLocked()) throw BluePrintProcessorException("failed to lock($lockName)")