Message Prioritization message group lock.

Implementation to avoid concurrent procession of message group while prioritization.

Sample message prioritization Kafka listener properties.

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

1 files changed, 7 insertions, 0 deletions

diff --git a/ms/blueprintsprocessor/modules/commons/atomix-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/atomix/service/AtomixBluePrintClusterService.kt b/ms/blueprintsprocessor/modules/commons/atomix-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/atomix/service/AtomixBluePrintClusterService.kt
index 0690eb89d..214a14310 100644
--- a/ms/blueprintsprocessor/modules/commons/atomix-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/atomix/service/AtomixBluePrintClusterService.kt
+++ b/ms/blueprintsprocessor/modules/commons/atomix-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/atomix/service/AtomixBluePrintClusterService.kt
@@ -148,12 +148,18 @@ open class AtomixBluePrintClusterService : BluePrintClusterService {
 }
 
 open class ClusterLockImpl(private val atomix: Atomix, private val name: String) : ClusterLock {
+    val log = logger(ClusterLockImpl::class)
 
     lateinit var distributedLock: DistributedLock
 
+    override fun name(): String {
+        return distributedLock.name()
+    }
+
     override suspend fun lock() {
         distributedLock = AtomixLibUtils.distributedLock(atomix, name)
         distributedLock.lock()
+        log.debug("Cluster lock($name) created..")
     }
 
     override suspend fun tryLock(timeout: Long): Boolean {
@@ -163,6 +169,7 @@ open class ClusterLockImpl(private val atomix: Atomix, private val name: String)
 
     override suspend fun unLock() {
         distributedLock.unlock()
+        log.debug("Cluster unlock(${name()}) successfully..")
     }
 
     override fun isLocked(): Boolean {