Formatting Code base with ktlint

No Business logic change, just the code format.
Competible with IntelliJ: https://github.com/pinterest/ktlint#option-3

To format run: mvn process-sources -P format

Issue-ID: CCSDK-1947
Signed-off-by: Singal, Kapil (ks220y) <ks220y@att.com>
Change-Id: Ic9e9209fb7023d77f434693ad5a01229f8d09331

21 files changed, 534 insertions, 365 deletions

diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/AbstractTopologyComponents.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/AbstractMessagePrioritizeProcessor.kt
index d114da521..c2965c4e8 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/AbstractTopologyComponents.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/AbstractMessagePrioritizeProcessor.kt
@@ -34,8 +34,6 @@ abstract class AbstractMessagePrioritizeProcessor<K, V> : AbstractBluePrintMessa
         this.processorContext = context
         /** Get the State service to update in store */
         this.messagePrioritizationStateService = BluePrintDependencyService
-                .messagePrioritizationStateService()
-
+            .messagePrioritizationStateService()
     }
-
-}
\ No newline at end of file
+}
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationConfiguration.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationConfiguration.kt
index cce883c91..28e096352 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationConfiguration.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationConfiguration.kt
@@ -23,7 +23,6 @@ import org.springframework.context.annotation.Configuration
 @ComponentScan
 open class MessagePrioritizationConfiguration
 
-
 object MessagePrioritizationConstants {
 
     const val SOURCE_INPUT = "source-prioritization-input"
@@ -34,4 +33,4 @@ object MessagePrioritizationConstants {
 
     const val SINK_OUTPUT = "sink-prioritization-output"
     const val SINK_EXPIRED = "sink-prioritization-expired"
-}
\ No newline at end of file
+}
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationConsumer.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationConsumer.kt
index 967cc190e..ed124d1b2 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationConsumer.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationConsumer.kt
@@ -28,7 +28,8 @@ import org.onap.ccsdk.cds.blueprintsprocessor.message.service.KafkaStreamConsume
 import org.onap.ccsdk.cds.controllerblueprints.core.logger
 
 open class MessagePrioritizationConsumer(
-        private val bluePrintMessageLibPropertyService: BluePrintMessageLibPropertyService) {
+    private val bluePrintMessageLibPropertyService: BluePrintMessageLibPropertyService
+) {
 
     private val log = logger(MessagePrioritizationConsumer::class)
 
@@ -36,15 +37,17 @@ open class MessagePrioritizationConsumer(
 
     open fun consumerService(selector: String): BlueprintMessageConsumerService {
         return bluePrintMessageLibPropertyService
-                .blueprintMessageConsumerService(selector)
+            .blueprintMessageConsumerService(selector)
     }
 
-    open fun kafkaStreamConsumerFunction(prioritizationConfiguration: PrioritizationConfiguration)
-            : KafkaStreamConsumerFunction {
+    open fun kafkaStreamConsumerFunction(prioritizationConfiguration: PrioritizationConfiguration):
+            KafkaStreamConsumerFunction {
         return object : KafkaStreamConsumerFunction {
 
-            override suspend fun createTopology(messageConsumerProperties: MessageConsumerProperties,
-                                                additionalConfig: Map<String, Any>?): Topology {
+            override suspend fun createTopology(
+                messageConsumerProperties: MessageConsumerProperties,
+                additionalConfig: Map<String, Any>?
+            ): Topology {
 
                 val topology = Topology()
                 val kafkaStreamsBasicAuthConsumerProperties = messageConsumerProperties
@@ -55,33 +58,49 @@ open class MessagePrioritizationConsumer(
 
                 topology.addSource(MessagePrioritizationConstants.SOURCE_INPUT, *topics.toTypedArray())
 
-                topology.addProcessor(MessagePrioritizationConstants.PROCESSOR_PRIORITIZE,
-                        bluePrintProcessorSupplier<ByteArray, ByteArray>(MessagePrioritizationConstants.PROCESSOR_PRIORITIZE,
-                                prioritizationConfiguration),
-                        MessagePrioritizationConstants.SOURCE_INPUT)
-
-                topology.addProcessor(MessagePrioritizationConstants.PROCESSOR_AGGREGATE,
-                        bluePrintProcessorSupplier<String, String>(MessagePrioritizationConstants.PROCESSOR_AGGREGATE,
-                                prioritizationConfiguration),
-                        MessagePrioritizationConstants.PROCESSOR_PRIORITIZE)
-
-                topology.addProcessor(MessagePrioritizationConstants.PROCESSOR_OUTPUT,
-                        bluePrintProcessorSupplier<String, String>(MessagePrioritizationConstants.PROCESSOR_OUTPUT,
-                                prioritizationConfiguration),
-                        MessagePrioritizationConstants.PROCESSOR_AGGREGATE)
-
-                topology.addSink(MessagePrioritizationConstants.SINK_EXPIRED,
-                        prioritizationConfiguration.expiredTopic,
-                        Serdes.String().serializer(), MessagePrioritizationSerde().serializer(),
-                        MessagePrioritizationConstants.PROCESSOR_PRIORITIZE)
-
-                /** To receive completed and error messages */
-                topology.addSink(MessagePrioritizationConstants.SINK_OUTPUT,
-                        prioritizationConfiguration.outputTopic,
-                        Serdes.String().serializer(), MessagePrioritizationSerde().serializer(),
+                topology.addProcessor(
+                    MessagePrioritizationConstants.PROCESSOR_PRIORITIZE,
+                    bluePrintProcessorSupplier<ByteArray, ByteArray>(
                         MessagePrioritizationConstants.PROCESSOR_PRIORITIZE,
+                        prioritizationConfiguration
+                    ),
+                    MessagePrioritizationConstants.SOURCE_INPUT
+                )
+
+                topology.addProcessor(
+                    MessagePrioritizationConstants.PROCESSOR_AGGREGATE,
+                    bluePrintProcessorSupplier<String, String>(
                         MessagePrioritizationConstants.PROCESSOR_AGGREGATE,
-                        MessagePrioritizationConstants.PROCESSOR_OUTPUT)
+                        prioritizationConfiguration
+                    ),
+                    MessagePrioritizationConstants.PROCESSOR_PRIORITIZE
+                )
+
+                topology.addProcessor(
+                    MessagePrioritizationConstants.PROCESSOR_OUTPUT,
+                    bluePrintProcessorSupplier<String, String>(
+                        MessagePrioritizationConstants.PROCESSOR_OUTPUT,
+                        prioritizationConfiguration
+                    ),
+                    MessagePrioritizationConstants.PROCESSOR_AGGREGATE
+                )
+
+                topology.addSink(
+                    MessagePrioritizationConstants.SINK_EXPIRED,
+                    prioritizationConfiguration.expiredTopic,
+                    Serdes.String().serializer(), MessagePrioritizationSerde().serializer(),
+                    MessagePrioritizationConstants.PROCESSOR_PRIORITIZE
+                )
+
+                /** To receive completed and error messages */
+                topology.addSink(
+                    MessagePrioritizationConstants.SINK_OUTPUT,
+                    prioritizationConfiguration.outputTopic,
+                    Serdes.String().serializer(), MessagePrioritizationSerde().serializer(),
+                    MessagePrioritizationConstants.PROCESSOR_PRIORITIZE,
+                    MessagePrioritizationConstants.PROCESSOR_AGGREGATE,
+                    MessagePrioritizationConstants.PROCESSOR_OUTPUT
+                )
 
                 // Output will be sent to the group-output topic from Processor API
                 return topology
@@ -102,4 +121,4 @@ open class MessagePrioritizationConsumer(
             streamingConsumerService.shutDown()
         }
     }
-}
\ No newline at end of file
+}
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationData.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationData.kt
index 3358a5643..3ecfa27e0 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationData.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationData.kt
@@ -61,8 +61,9 @@ open class UpdateStateRequest : Serializable {
     var state: String? = null
 }
 
-data class CorrelationCheckResponse(var message: String? = null,
-                                    var correlated: Boolean = false)
+data class CorrelationCheckResponse(
+    var message: String? = null,
+    var correlated: Boolean = false
+)
 
 data class TypeCorrelationKey(val type: String, val correlationId: String)
-
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 ec061ad47..39d081455 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
@@ -21,30 +21,29 @@ import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.s
 import org.onap.ccsdk.cds.blueprintsprocessor.services.execution.AbstractComponentFunction
 import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService
 
-
 /**
  * Register the MessagePrioritizationStateService and exposed dependency
  */
 fun BluePrintDependencyService.messagePrioritizationStateService(): MessagePrioritizationStateService =
-        instance(MessagePrioritizationStateService::class)
+    instance(MessagePrioritizationStateService::class)
 
 /**
  * Expose messagePrioritizationStateService to AbstractComponentFunction
  */
 fun AbstractComponentFunction.messagePrioritizationStateService() =
-        BluePrintDependencyService.messagePrioritizationStateService()
+    BluePrintDependencyService.messagePrioritizationStateService()
 
 /**
  * MessagePrioritization correlation extensions
  */
 fun MessagePrioritization.toFormatedCorrelation(): String {
     val ascendingKey = this.correlationId!!.split(",")
-            .map { it.trim() }.sorted().joinToString(",")
+        .map { it.trim() }.sorted().joinToString(",")
     return ascendingKey
 }
 
 fun MessagePrioritization.toTypeNCorrelation(): TypeCorrelationKey {
     val ascendingKey = this.correlationId!!.split(",")
-            .map { it.trim() }.sorted().joinToString(",")
+        .map { it.trim() }.sorted().joinToString(",")
     return TypeCorrelationKey(this.type, ascendingKey)
 }
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/api/MessagePrioritizationApi.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/api/MessagePrioritizationApi.kt
index 382cb9c8a..262dcb402 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/api/MessagePrioritizationApi.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/api/MessagePrioritizationApi.kt
@@ -21,7 +21,13 @@ import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.d
 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.service.MessagePrioritizationStateService
 import org.onap.ccsdk.cds.blueprintsprocessor.rest.service.monoMdc
 import org.springframework.http.MediaType
-import org.springframework.web.bind.annotation.*
+import org.springframework.web.bind.annotation.GetMapping
+import org.springframework.web.bind.annotation.PathVariable
+import org.springframework.web.bind.annotation.PostMapping
+import org.springframework.web.bind.annotation.RequestBody
+import org.springframework.web.bind.annotation.RequestMapping
+import org.springframework.web.bind.annotation.ResponseBody
+import org.springframework.web.bind.annotation.RestController
 
 @RestController
 @RequestMapping(value = ["/api/v1/message-prioritization"])
@@ -31,25 +37,30 @@ open class MessagePrioritizationApi(private val messagePrioritizationStateServic
     @ResponseBody
     fun ping(): String = "Success"
 
-
     @GetMapping(path = ["/{id}"], produces = [MediaType.APPLICATION_JSON_VALUE])
     @ResponseBody
     fun messagePrioritization(@PathVariable(value = "id") id: String) = monoMdc {
         messagePrioritizationStateService.getMessage(id)
     }
 
-    @PostMapping(path = ["/"], produces = [MediaType.APPLICATION_JSON_VALUE],
-            consumes = [MediaType.APPLICATION_JSON_VALUE])
+    @PostMapping(
+        path = ["/"], produces = [MediaType.APPLICATION_JSON_VALUE],
+        consumes = [MediaType.APPLICATION_JSON_VALUE]
+    )
     @ResponseBody
     fun saveMessagePrioritization(@RequestBody messagePrioritization: MessagePrioritization) = monoMdc {
         messagePrioritizationStateService.saveMessage(messagePrioritization)
     }
 
-    @PostMapping(path = ["/update-state"], produces = [MediaType.APPLICATION_JSON_VALUE],
-            consumes = [MediaType.APPLICATION_JSON_VALUE])
+    @PostMapping(
+        path = ["/update-state"], produces = [MediaType.APPLICATION_JSON_VALUE],
+        consumes = [MediaType.APPLICATION_JSON_VALUE]
+    )
     fun updateMessagePrioritizationState(@RequestBody updateMessageState: UpdateStateRequest) =
-            monoMdc {
-                messagePrioritizationStateService.setMessageState(updateMessageState.id,
-                        updateMessageState.state!!)
-            }
-}
\ No newline at end of file
+        monoMdc {
+            messagePrioritizationStateService.setMessageState(
+                updateMessageState.id,
+                updateMessageState.state!!
+            )
+        }
+}
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/db/PrioritizationMessageEntity.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/db/MessagePrioritization.kt
index 15e85b0e7..ce2085f68 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/db/PrioritizationMessageEntity.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/db/MessagePrioritization.kt
@@ -21,8 +21,15 @@ import org.hibernate.annotations.Proxy
 import org.springframework.data.annotation.LastModifiedDate
 import org.springframework.data.jpa.domain.support.AuditingEntityListener
 import org.springframework.data.jpa.repository.config.EnableJpaAuditing
-import java.util.*
-import javax.persistence.*
+import java.util.Date
+import javax.persistence.Column
+import javax.persistence.Entity
+import javax.persistence.EntityListeners
+import javax.persistence.Id
+import javax.persistence.Lob
+import javax.persistence.Table
+import javax.persistence.Temporal
+import javax.persistence.TemporalType
 
 @EnableJpaAuditing
 @EntityListeners(AuditingEntityListener::class)
@@ -30,6 +37,7 @@ import javax.persistence.*
 @Table(name = "MESSAGE_PRIORITIZATION")
 @Proxy(lazy = false)
 open class MessagePrioritization {
+
     @Id
     @Column(name = "message_id", length = 50)
     lateinit var id: String
@@ -78,4 +86,4 @@ open class MessagePrioritization {
     @Temporal(TemporalType.TIMESTAMP)
     @Column(name = "expiry_date", nullable = false)
     var expiryDate: Date? = null
-}
\ No newline at end of file
+}
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/db/MessagePrioritizationRepositories.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/db/MessagePrioritizationRepositories.kt
deleted file mode 100644
index 5c2495fd7..000000000
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/db/MessagePrioritizationRepositories.kt
+++ /dev/null
@@ -1,118 +0,0 @@
-/*
- * Copyright © 2018-2019 AT&T Intellectual Property.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db
-
-import org.springframework.data.domain.Pageable
-import org.springframework.data.jpa.repository.JpaRepository
-import org.springframework.data.jpa.repository.Modifying
-import org.springframework.data.jpa.repository.Query
-import org.springframework.stereotype.Repository
-import org.springframework.transaction.annotation.Transactional
-import java.util.*
-
-@Repository
-@Transactional(readOnly = true)
-interface PrioritizationMessageRepository : JpaRepository<MessagePrioritization, String> {
-
-    @Query("FROM MessagePrioritization pm WHERE pm.group = :group ORDER BY pm.createdDate asc")
-    fun findByGroup(group: String, count: Pageable): List<MessagePrioritization>?
-
-    @Query("FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state in :states " +
-            "ORDER BY pm.createdDate asc")
-    fun findByGroupAndStateIn(group: String, states: List<String>, count: Pageable): List<MessagePrioritization>?
-
-    @Query("FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state in :states " +
-            "ORDER BY pm.updatedDate asc")
-    fun findByGroupAndStateInOrderByUpdatedDate(group: String, states: List<String>, count: Pageable)
-            : List<MessagePrioritization>?
-
-    @Query("FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state in :states " +
-            "AND pm.expiryDate > :expiryCheckDate ORDER BY pm.createdDate asc")
-    fun findByGroupAndStateInAndNotExpiredDate(group: String, states: List<String>, expiryCheckDate: Date,
-                                               count: Pageable): List<MessagePrioritization>?
-
-    @Query("FROM MessagePrioritization pm WHERE pm.state in :states " +
-            "AND pm.expiryDate < :expiryCheckDate ORDER BY pm.createdDate asc")
-    fun findByStateInAndExpiredDate(states: List<String>, expiryCheckDate: Date,
-                                    count: Pageable): List<MessagePrioritization>?
-
-    @Query("FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state in :states " +
-            "AND pm.expiryDate < :expiryCheckDate ORDER BY pm.createdDate asc")
-    fun findByGroupAndStateInAndExpiredDate(group: String, states: List<String>, expiryCheckDate: Date,
-                                            count: Pageable): List<MessagePrioritization>?
-
-    @Query("FROM MessagePrioritization pm WHERE pm.group = :group " +
-            "AND pm.expiryDate < :expiryCheckDate ORDER BY pm.createdDate asc")
-    fun findByByGroupAndExpiredDate(group: String, expiryCheckDate: Date, count: Pageable): List<MessagePrioritization>?
-
-    @Query("FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state in :states " +
-            "AND pm.correlationId = :correlationId ORDER BY pm.createdDate asc")
-    fun findByGroupAndCorrelationId(group: String, states: List<String>, correlationId: String)
-            : List<MessagePrioritization>?
-
-    @Query("FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state in :states " +
-            "AND pm.type in :types AND pm.correlationId = :correlationId ORDER BY pm.createdDate asc")
-    fun findByGroupAndTypesAndCorrelationId(group: String, states: List<String>, types: List<String>,
-                                            correlationId: String): List<MessagePrioritization>?
-
-    @Modifying
-    @Transactional
-    @Query("UPDATE MessagePrioritization SET state = :state, updatedDate = :currentDate " +
-            "WHERE id = :id")
-    fun setStateForMessageId(id: String, state: String, currentDate: Date): Int
-
-    @Modifying
-    @Transactional
-    @Query("UPDATE MessagePrioritization SET priority = :priority, updatedDate = :currentDate " +
-            "WHERE id = :id")
-    fun setPriorityForMessageId(id: String, priority: String, currentDate: Date): Int
-
-    @Modifying
-    @Transactional
-    @Query("UPDATE MessagePrioritization SET state = :state, updatedDate = :currentDate " +
-            "WHERE id IN :ids")
-    fun setStateForMessageIds(ids: List<String>, state: String, currentDate: Date): Int
-
-    @Modifying
-    @Transactional
-    @Query("UPDATE MessagePrioritization SET priority = :priority, updatedDate = :currentDate " +
-            "WHERE id IN :ids")
-    fun setPriorityForMessageIds(ids: List<String>, priority: String, currentDate: Date): Int
-
-    @Modifying
-    @Transactional
-    @Query("UPDATE MessagePrioritization SET state = :state, error = :error, updatedDate = :currentDate " +
-            "WHERE id = :id")
-    fun setStateAndErrorForMessageId(id: String, state: String, error: String, currentDate: Date): Int
-
-    @Modifying
-    @Transactional
-    @Query("UPDATE MessagePrioritization SET state = :state, " +
-            "aggregatedMessageIds = :aggregatedMessageIds, updatedDate = :currentDate WHERE id = :id")
-    fun setStateAndAggregatedMessageIds(id: String, state: String, aggregatedMessageIds: String, currentDate: Date): Int
-
-    @Modifying
-    @Transactional
-    @Query("DELETE FROM MessagePrioritization pm WHERE pm.group = :group")
-    fun deleteGroup(group: String)
-
-    @Modifying
-    @Transactional
-    @Query("DELETE FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state IN :states")
-    fun deleteGroupAndStateIn(group: String, states: List<String>)
-}
-
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/db/PrioritizationMessageRepository.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/db/PrioritizationMessageRepository.kt
new file mode 100644
index 000000000..b0514838a
--- /dev/null
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/db/PrioritizationMessageRepository.kt
@@ -0,0 +1,160 @@
+/*
+ * Copyright © 2018-2019 AT&T Intellectual Property.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db
+
+import org.springframework.data.domain.Pageable
+import org.springframework.data.jpa.repository.JpaRepository
+import org.springframework.data.jpa.repository.Modifying
+import org.springframework.data.jpa.repository.Query
+import org.springframework.stereotype.Repository
+import org.springframework.transaction.annotation.Transactional
+import java.util.Date
+
+@Repository
+@Transactional(readOnly = true)
+interface PrioritizationMessageRepository : JpaRepository<MessagePrioritization, String> {
+
+    @Query("FROM MessagePrioritization pm WHERE pm.group = :group ORDER BY pm.createdDate asc")
+    fun findByGroup(group: String, count: Pageable): List<MessagePrioritization>?
+
+    @Query(
+        "FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state in :states " +
+                "ORDER BY pm.createdDate asc"
+    )
+    fun findByGroupAndStateIn(group: String, states: List<String>, count: Pageable): List<MessagePrioritization>?
+
+    @Query(
+        "FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state in :states " +
+                "ORDER BY pm.updatedDate asc"
+    )
+    fun findByGroupAndStateInOrderByUpdatedDate(group: String, states: List<String>, count: Pageable):
+            List<MessagePrioritization>?
+
+    @Query(
+        "FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state in :states " +
+                "AND pm.expiryDate > :expiryCheckDate ORDER BY pm.createdDate asc"
+    )
+    fun findByGroupAndStateInAndNotExpiredDate(
+        group: String,
+        states: List<String>,
+        expiryCheckDate: Date,
+        count: Pageable
+    ): List<MessagePrioritization>?
+
+    @Query(
+        "FROM MessagePrioritization pm WHERE pm.state in :states " +
+                "AND pm.expiryDate < :expiryCheckDate ORDER BY pm.createdDate asc"
+    )
+    fun findByStateInAndExpiredDate(
+        states: List<String>,
+        expiryCheckDate: Date,
+        count: Pageable
+    ): List<MessagePrioritization>?
+
+    @Query(
+        "FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state in :states " +
+                "AND pm.expiryDate < :expiryCheckDate ORDER BY pm.createdDate asc"
+    )
+    fun findByGroupAndStateInAndExpiredDate(
+        group: String,
+        states: List<String>,
+        expiryCheckDate: Date,
+        count: Pageable
+    ): List<MessagePrioritization>?
+
+    @Query(
+        "FROM MessagePrioritization pm WHERE pm.group = :group " +
+                "AND pm.expiryDate < :expiryCheckDate ORDER BY pm.createdDate asc"
+    )
+    fun findByByGroupAndExpiredDate(group: String, expiryCheckDate: Date, count: Pageable): List<MessagePrioritization>?
+
+    @Query(
+        "FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state in :states " +
+                "AND pm.correlationId = :correlationId ORDER BY pm.createdDate asc"
+    )
+    fun findByGroupAndCorrelationId(group: String, states: List<String>, correlationId: String):
+            List<MessagePrioritization>?
+
+    @Query(
+        "FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state in :states " +
+                "AND pm.type in :types AND pm.correlationId = :correlationId ORDER BY pm.createdDate asc"
+    )
+    fun findByGroupAndTypesAndCorrelationId(
+        group: String,
+        states: List<String>,
+        types: List<String>,
+        correlationId: String
+    ): List<MessagePrioritization>?
+
+    @Modifying
+    @Transactional
+    @Query(
+        "UPDATE MessagePrioritization SET state = :state, updatedDate = :currentDate " +
+                "WHERE id = :id"
+    )
+    fun setStateForMessageId(id: String, state: String, currentDate: Date): Int
+
+    @Modifying
+    @Transactional
+    @Query(
+        "UPDATE MessagePrioritization SET priority = :priority, updatedDate = :currentDate " +
+                "WHERE id = :id"
+    )
+    fun setPriorityForMessageId(id: String, priority: String, currentDate: Date): Int
+
+    @Modifying
+    @Transactional
+    @Query(
+        "UPDATE MessagePrioritization SET state = :state, updatedDate = :currentDate " +
+                "WHERE id IN :ids"
+    )
+    fun setStateForMessageIds(ids: List<String>, state: String, currentDate: Date): Int
+
+    @Modifying
+    @Transactional
+    @Query(
+        "UPDATE MessagePrioritization SET priority = :priority, updatedDate = :currentDate " +
+                "WHERE id IN :ids"
+    )
+    fun setPriorityForMessageIds(ids: List<String>, priority: String, currentDate: Date): Int
+
+    @Modifying
+    @Transactional
+    @Query(
+        "UPDATE MessagePrioritization SET state = :state, error = :error, updatedDate = :currentDate " +
+                "WHERE id = :id"
+    )
+    fun setStateAndErrorForMessageId(id: String, state: String, error: String, currentDate: Date): Int
+
+    @Modifying
+    @Transactional
+    @Query(
+        "UPDATE MessagePrioritization SET state = :state, " +
+                "aggregatedMessageIds = :aggregatedMessageIds, updatedDate = :currentDate WHERE id = :id"
+    )
+    fun setStateAndAggregatedMessageIds(id: String, state: String, aggregatedMessageIds: String, currentDate: Date): Int
+
+    @Modifying
+    @Transactional
+    @Query("DELETE FROM MessagePrioritization pm WHERE pm.group = :group")
+    fun deleteGroup(group: String)
+
+    @Modifying
+    @Transactional
+    @Query("DELETE FROM MessagePrioritization pm WHERE pm.group = :group AND pm.state IN :states")
+    fun deleteGroupAndStateIn(group: String, states: List<String>)
+}
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/service/MessagePrioritizationStateService.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/service/MessagePrioritizationStateService.kt
index 6138fa9d3..017658ff6 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/service/MessagePrioritizationStateService.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/service/MessagePrioritizationStateService.kt
@@ -25,7 +25,7 @@ import org.onap.ccsdk.cds.controllerblueprints.core.logger
 import org.springframework.data.domain.PageRequest
 import org.springframework.stereotype.Service
 import org.springframework.transaction.annotation.Transactional
-import java.util.*
+import java.util.Date
 
 interface MessagePrioritizationStateService {
 
@@ -37,16 +37,20 @@ interface MessagePrioritizationStateService {
 
     suspend fun getExpiryEligibleMessages(count: Int): List<MessagePrioritization>?
 
-    suspend fun getMessageForStatesNotExpiredIn(group: String, states: List<String>, count: Int)
-            : List<MessagePrioritization>?
+    suspend fun getMessageForStatesNotExpiredIn(group: String, states: List<String>, count: Int):
+            List<MessagePrioritization>?
 
-    suspend fun getMessageForStatesExpired(group: String, states: List<String>, count: Int)
-            : List<MessagePrioritization>?
+    suspend fun getMessageForStatesExpired(group: String, states: List<String>, count: Int):
+            List<MessagePrioritization>?
 
     suspend fun getExpiredMessages(group: String, expiryDate: Date, count: Int): List<MessagePrioritization>?
 
-    suspend fun getCorrelatedMessages(group: String, states: List<String>, types: List<String>?,
-                                      correlationIds: String): List<MessagePrioritization>?
+    suspend fun getCorrelatedMessages(
+        group: String,
+        states: List<String>,
+        types: List<String>?,
+        correlationIds: String
+    ): List<MessagePrioritization>?
 
     suspend fun updateMessagesState(ids: List<String>, state: String)
 
@@ -73,8 +77,9 @@ interface MessagePrioritizationStateService {
 
 @Service
 open class MessagePrioritizationStateServiceImpl(
-        private val prioritizationMessageRepository: PrioritizationMessageRepository)
-    : MessagePrioritizationStateService {
+    private val prioritizationMessageRepository: PrioritizationMessageRepository
+) :
+    MessagePrioritizationStateService {
 
     private val log = logger(MessagePrioritizationStateServiceImpl::class)
 
@@ -89,7 +94,7 @@ open class MessagePrioritizationStateServiceImpl(
 
     override suspend fun getMessage(id: String): MessagePrioritization {
         return prioritizationMessageRepository.findById(id).orElseGet(null)
-                ?: throw BluePrintProcessorException("couldn't find message for id($id)")
+            ?: throw BluePrintProcessorException("couldn't find message for id($id)")
     }
 
     override suspend fun getMessages(ids: List<String>): List<MessagePrioritization>? {
@@ -98,30 +103,42 @@ open class MessagePrioritizationStateServiceImpl(
 
     override suspend fun getExpiryEligibleMessages(count: Int): List<MessagePrioritization>? {
         return prioritizationMessageRepository
-                .findByStateInAndExpiredDate(arrayListOf(MessageState.NEW.name, MessageState.WAIT.name),
-                        Date(), PageRequest.of(0, count))
-    }
-
-    override suspend fun getMessageForStatesNotExpiredIn(group: String, states: List<String>, count: Int)
-            : List<MessagePrioritization>? {
-        return prioritizationMessageRepository.findByGroupAndStateInAndNotExpiredDate(group,
-                states, Date(), PageRequest.of(0, count))
-    }
-
-    override suspend fun getMessageForStatesExpired(group: String, states: List<String>, count: Int)
-            : List<MessagePrioritization>? {
-        return prioritizationMessageRepository.findByGroupAndStateInAndExpiredDate(group,
-                states, Date(), PageRequest.of(0, count))
-    }
-
-    override suspend fun getExpiredMessages(group: String, expiryDate: Date, count: Int)
-            : List<MessagePrioritization>? {
-        return prioritizationMessageRepository.findByByGroupAndExpiredDate(group,
-                expiryDate, PageRequest.of(0, count))
-    }
-
-    override suspend fun getCorrelatedMessages(group: String, states: List<String>, types: List<String>?,
-                                               correlationIds: String): List<MessagePrioritization>? {
+            .findByStateInAndExpiredDate(
+                arrayListOf(MessageState.NEW.name, MessageState.WAIT.name),
+                Date(), PageRequest.of(0, count)
+            )
+    }
+
+    override suspend fun getMessageForStatesNotExpiredIn(group: String, states: List<String>, count: Int):
+            List<MessagePrioritization>? {
+        return prioritizationMessageRepository.findByGroupAndStateInAndNotExpiredDate(
+            group,
+            states, Date(), PageRequest.of(0, count)
+        )
+    }
+
+    override suspend fun getMessageForStatesExpired(group: String, states: List<String>, count: Int):
+            List<MessagePrioritization>? {
+        return prioritizationMessageRepository.findByGroupAndStateInAndExpiredDate(
+            group,
+            states, Date(), PageRequest.of(0, count)
+        )
+    }
+
+    override suspend fun getExpiredMessages(group: String, expiryDate: Date, count: Int):
+            List<MessagePrioritization>? {
+        return prioritizationMessageRepository.findByByGroupAndExpiredDate(
+            group,
+            expiryDate, PageRequest.of(0, count)
+        )
+    }
+
+    override suspend fun getCorrelatedMessages(
+        group: String,
+        states: List<String>,
+        types: List<String>?,
+        correlationIds: String
+    ): List<MessagePrioritization>? {
         return if (!types.isNullOrEmpty()) {
             prioritizationMessageRepository.findByGroupAndTypesAndCorrelationId(group, states, types, correlationIds)
         } else {
@@ -185,7 +202,9 @@ open class MessagePrioritizationStateServiceImpl(
     }
 
     override suspend fun deleteExpiredMessage(group: String, retentionDays: Int) {
-        return prioritizationMessageRepository.deleteGroupAndStateIn(group,
-                arrayListOf(MessageState.EXPIRED.name))
+        return prioritizationMessageRepository.deleteGroupAndStateIn(
+            group,
+            arrayListOf(MessageState.EXPIRED.name)
+        )
     }
-}
\ No newline at end of file
+}
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessageAggregateProcessor.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessageAggregateProcessor.kt
index 45f5c773d..3e697e633 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessageAggregateProcessor.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessageAggregateProcessor.kt
@@ -22,7 +22,6 @@ import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.M
 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessageState
 import org.onap.ccsdk.cds.controllerblueprints.core.logger
 
-
 open class MessageAggregateProcessor : AbstractMessagePrioritizeProcessor<String, String>() {
 
     private val log = logger(MessageAggregateProcessor::class)
@@ -50,16 +49,21 @@ open class MessageAggregateProcessor : AbstractMessagePrioritizeProcessor<String
                     storeMessages.forEach { messagePrioritization ->
                         try {
                             /** Update the data store */
-                            messagePrioritizationStateService.setMessageStateANdError(messagePrioritization.id,
-                                    MessageState.ERROR.name, error)
+                            messagePrioritizationStateService.setMessageStateANdError(
+                                messagePrioritization.id,
+                                MessageState.ERROR.name, error
+                            )
                             /** Publish to Error topic */
-                            this.processorContext.forward(messagePrioritization.id, messagePrioritization,
-                                    To.child(MessagePrioritizationConstants.SINK_OUTPUT))
+                            this.processorContext.forward(
+                                messagePrioritization.id, messagePrioritization,
+                                To.child(MessagePrioritizationConstants.SINK_OUTPUT)
+                            )
                         } catch (sendException: Exception) {
-                            log.error("failed to update/publish error message(${messagePrioritization.id}) : " +
-                                    "${sendException.message}", e)
+                            log.error(
+                                "failed to update/publish error message(${messagePrioritization.id}) : " +
+                                        "${sendException.message}", e
+                            )
                         }
-
                     }
                 }
             }
@@ -73,4 +77,4 @@ open class MessageAggregateProcessor : AbstractMessagePrioritizeProcessor<String
             processorContext.forward(id, id, To.child(MessagePrioritizationConstants.PROCESSOR_OUTPUT))
         }
     }
-}
\ No newline at end of file
+}
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessageOutputProcessor.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessageOutputProcessor.kt
index 34faa1b3b..cf6520df5 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessageOutputProcessor.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessageOutputProcessor.kt
@@ -22,7 +22,6 @@ import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.M
 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessageState
 import org.onap.ccsdk.cds.controllerblueprints.core.logger
 
-
 open class MessageOutputProcessor : AbstractMessagePrioritizeProcessor<String, String>() {
 
     private val log = logger(MessageOutputProcessor::class)
@@ -32,4 +31,4 @@ open class MessageOutputProcessor : AbstractMessagePrioritizeProcessor<String, S
         val message = messagePrioritizationStateService.updateMessageState(value, MessageState.COMPLETED.name)
         processorContext.forward(message.id, message, To.child(MessagePrioritizationConstants.SINK_OUTPUT))
     }
-}
\ No newline at end of file
+}
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessagePrioritizationPunctuators.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessagePrioritizationPunctuators.kt
index a745e034c..5435ebe30 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessagePrioritizationPunctuators.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessagePrioritizationPunctuators.kt
@@ -24,41 +24,46 @@ import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.s
 import org.onap.ccsdk.cds.blueprintsprocessor.message.kafka.AbstractBluePrintMessagePunctuator
 import org.onap.ccsdk.cds.controllerblueprints.core.logger
 
-
-class MessagePriorityExpiryPunctuator(private val messagePrioritizationStateService: MessagePrioritizationStateService)
-    : AbstractBluePrintMessagePunctuator() {
+class MessagePriorityExpiryPunctuator(private val messagePrioritizationStateService: MessagePrioritizationStateService) :
+    AbstractBluePrintMessagePunctuator() {
 
     private val log = logger(MessagePriorityExpiryPunctuator::class)
     lateinit var configuration: PrioritizationConfiguration
 
     override suspend fun punctuateNB(timestamp: Long) {
 
-        log.info("**** executing expiry punctuator applicationId(${processorContext.applicationId()}), " +
-                "taskId(${processorContext.taskId()})")
+        log.info(
+            "**** executing expiry punctuator applicationId(${processorContext.applicationId()}), " +
+                    "taskId(${processorContext.taskId()})"
+        )
         val expiryConfiguration = configuration.expiryConfiguration
         val fetchMessages = messagePrioritizationStateService
-                .getExpiryEligibleMessages(expiryConfiguration.maxPollRecord)
+            .getExpiryEligibleMessages(expiryConfiguration.maxPollRecord)
 
         val expiredIds = fetchMessages?.map { it.id }
         if (expiredIds != null && expiredIds.isNotEmpty()) {
             messagePrioritizationStateService.updateMessagesState(expiredIds, MessageState.EXPIRED.name)
             fetchMessages.forEach { expired ->
-                processorContext.forward(expired.id, expired,
-                        To.child(MessagePrioritizationConstants.SINK_EXPIRED))
+                processorContext.forward(
+                    expired.id, expired,
+                    To.child(MessagePrioritizationConstants.SINK_EXPIRED)
+                )
             }
         }
     }
 }
 
-class MessagePriorityCleanPunctuator(private val messagePrioritizationStateService: MessagePrioritizationStateService)
-    : AbstractBluePrintMessagePunctuator() {
+class MessagePriorityCleanPunctuator(private val messagePrioritizationStateService: MessagePrioritizationStateService) :
+    AbstractBluePrintMessagePunctuator() {
 
     private val log = logger(MessagePriorityCleanPunctuator::class)
     lateinit var configuration: PrioritizationConfiguration
 
     override suspend fun punctuateNB(timestamp: Long) {
-        log.info("**** executing clean punctuator applicationId(${processorContext.applicationId()}), " +
-                "taskId(${processorContext.taskId()})")
-        //TODO
+        log.info(
+            "**** executing clean punctuator applicationId(${processorContext.applicationId()}), " +
+                    "taskId(${processorContext.taskId()})"
+        )
+        // TODO
     }
-}
\ No newline at end of file
+}
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessagePrioritizationSerdes.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessagePrioritizationSerde.kt
index 00d454727..f2a481f74 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessagePrioritizationSerdes.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessagePrioritizationSerde.kt
@@ -37,7 +37,7 @@ open class MessagePrioritizationSerde : Serde<MessagePrioritization> {
         return object : Deserializer<MessagePrioritization> {
             override fun deserialize(topic: String, data: ByteArray): MessagePrioritization {
                 return JacksonUtils.readValue(String(data), MessagePrioritization::class.java)
-                        ?: throw BluePrintProcessorException("failed to convert")
+                    ?: throw BluePrintProcessorException("failed to convert")
             }
 
             override fun configure(configs: MutableMap<String, *>?, isKey: Boolean) {
@@ -61,4 +61,4 @@ open class MessagePrioritizationSerde : Serde<MessagePrioritization> {
             }
         }
     }
-}
\ No newline at end of file
+}
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessagePrioritizeProcessor.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessagePrioritizeProcessor.kt
index 7dde2655d..431e02f30 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessagePrioritizeProcessor.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessagePrioritizeProcessor.kt
@@ -29,8 +29,7 @@ import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
 import org.onap.ccsdk.cds.controllerblueprints.core.logger
 import org.onap.ccsdk.cds.controllerblueprints.core.utils.JacksonUtils
 import java.time.Duration
-import java.util.*
-
+import java.util.UUID
 
 open class MessagePrioritizeProcessor : AbstractMessagePrioritizeProcessor<ByteArray, ByteArray>() {
 
@@ -42,7 +41,7 @@ open class MessagePrioritizeProcessor : AbstractMessagePrioritizeProcessor<ByteA
     override suspend fun processNB(key: ByteArray, value: ByteArray) {
         log.info("***** received in prioritize processor key(${String(key)})")
         val messagePrioritize = JacksonUtils.readValue(String(value), MessagePrioritization::class.java)
-                ?: throw BluePrintProcessorException("failed to convert")
+            ?: throw BluePrintProcessorException("failed to convert")
         try {
             // Save the Message
             messagePrioritizationStateService.saveMessage(messagePrioritize)
@@ -51,11 +50,15 @@ open class MessagePrioritizeProcessor : AbstractMessagePrioritizeProcessor<ByteA
             messagePrioritize.error = "failed in Prioritize message(${messagePrioritize.id}) : ${e.message}"
             log.error(messagePrioritize.error)
             /** Update the data store */
-            messagePrioritizationStateService.setMessageStateANdError(messagePrioritize.id, MessageState.ERROR.name,
-                    messagePrioritize.error!!)
+            messagePrioritizationStateService.setMessageStateANdError(
+                messagePrioritize.id, MessageState.ERROR.name,
+                messagePrioritize.error!!
+            )
             /** Publish to Output topic */
-            this.processorContext.forward(messagePrioritize.id, messagePrioritize,
-                    To.child(MessagePrioritizationConstants.SINK_OUTPUT))
+            this.processorContext.forward(
+                messagePrioritize.id, messagePrioritize,
+                To.child(MessagePrioritizationConstants.SINK_OUTPUT)
+            )
         }
     }
 
@@ -68,8 +71,10 @@ open class MessagePrioritizeProcessor : AbstractMessagePrioritizeProcessor<ByteA
     }
 
     override fun close() {
-        log.info("closing prioritization processor applicationId(${processorContext.applicationId()}), " +
-                "taskId(${processorContext.taskId()})")
+        log.info(
+            "closing prioritization processor applicationId(${processorContext.applicationId()}), " +
+                    "taskId(${processorContext.taskId()})"
+        )
         expiryCancellable.cancel()
         cleanCancellable.cancel()
     }
@@ -79,8 +84,10 @@ open class MessagePrioritizeProcessor : AbstractMessagePrioritizeProcessor<ByteA
         expiryPunctuator.processorContext = processorContext
         expiryPunctuator.configuration = prioritizationConfiguration
         val expiryConfiguration = prioritizationConfiguration.expiryConfiguration
-        expiryCancellable = processorContext.schedule(Duration.ofMillis(expiryConfiguration.frequencyMilli),
-                PunctuationType.WALL_CLOCK_TIME, expiryPunctuator)
+        expiryCancellable = processorContext.schedule(
+            Duration.ofMillis(expiryConfiguration.frequencyMilli),
+            PunctuationType.WALL_CLOCK_TIME, expiryPunctuator
+        )
         log.info("Expiry punctuator setup complete with frequency(${expiryConfiguration.frequencyMilli})mSec")
     }
 
@@ -89,10 +96,14 @@ open class MessagePrioritizeProcessor : AbstractMessagePrioritizeProcessor<ByteA
         cleanPunctuator.processorContext = processorContext
         cleanPunctuator.configuration = prioritizationConfiguration
         val cleanConfiguration = prioritizationConfiguration.cleanConfiguration
-        cleanCancellable = processorContext.schedule(Duration.ofDays(cleanConfiguration.expiredRecordsHoldDays.toLong()),
-                PunctuationType.WALL_CLOCK_TIME, cleanPunctuator)
-        log.info("Clean punctuator setup complete with expiry " +
-                "hold(${cleanConfiguration.expiredRecordsHoldDays})days")
+        cleanCancellable = processorContext.schedule(
+            Duration.ofDays(cleanConfiguration.expiredRecordsHoldDays.toLong()),
+            PunctuationType.WALL_CLOCK_TIME, cleanPunctuator
+        )
+        log.info(
+            "Clean punctuator setup complete with expiry " +
+                    "hold(${cleanConfiguration.expiredRecordsHoldDays})days"
+        )
     }
 
     open suspend fun handleCorrelationAndNextStep(messagePrioritization: MessagePrioritization) {
@@ -102,25 +113,31 @@ open class MessagePrioritizeProcessor : AbstractMessagePrioritizeProcessor<ByteA
             val group = messagePrioritization.group
             val correlationId = messagePrioritization.correlationId!!
             val types = getGroupCorrelationTypes(messagePrioritization)
-            log.info("checking correlation for message($id), group($group), types($types), " +
-                    "correlation id($correlationId)")
+            log.info(
+                "checking correlation for message($id), group($group), types($types), " +
+                        "correlation id($correlationId)"
+            )
 
             /** Get all previously received messages from database for group and optional types and correlation Id */
-            val waitingCorrelatedStoreMessages = messagePrioritizationStateService.getCorrelatedMessages(group,
-                    arrayListOf(MessageState.NEW.name, MessageState.WAIT.name), types, correlationId)
+            val waitingCorrelatedStoreMessages = messagePrioritizationStateService.getCorrelatedMessages(
+                group,
+                arrayListOf(MessageState.NEW.name, MessageState.WAIT.name), types, correlationId
+            )
 
             /** If multiple records found, then check correlation */
             if (!waitingCorrelatedStoreMessages.isNullOrEmpty() && waitingCorrelatedStoreMessages.size > 1) {
                 /** Check all correlation satisfies */
                 val correlationResults = MessageCorrelationUtils
-                        .correlatedMessagesWithTypes(waitingCorrelatedStoreMessages, types)
+                    .correlatedMessagesWithTypes(waitingCorrelatedStoreMessages, types)
 
                 if (correlationResults.correlated) {
                     /** Correlation  satisfied */
                     val correlatedIds = waitingCorrelatedStoreMessages.map { it.id }.joinToString(",")
                     /**  Send only correlated ids to next processor */
-                    this.processorContext.forward(UUID.randomUUID().toString(), correlatedIds,
-                            To.child(MessagePrioritizationConstants.PROCESSOR_AGGREGATE))
+                    this.processorContext.forward(
+                        UUID.randomUUID().toString(), correlatedIds,
+                        To.child(MessagePrioritizationConstants.PROCESSOR_AGGREGATE)
+                    )
                 } else {
                     /** Correlation not satisfied */
                     log.trace("correlation not matched : ${correlationResults.message}")
@@ -135,8 +152,10 @@ open class MessagePrioritizeProcessor : AbstractMessagePrioritizeProcessor<ByteA
         } else {
             // No Correlation check needed, simply forward to next processor.
             messagePrioritizationStateService.setMessageState(messagePrioritization.id, MessageState.PRIORITIZED.name)
-            this.processorContext.forward(messagePrioritization.id, messagePrioritization.id,
-                    To.child(MessagePrioritizationConstants.PROCESSOR_AGGREGATE))
+            this.processorContext.forward(
+                messagePrioritization.id, messagePrioritization.id,
+                To.child(MessagePrioritizationConstants.PROCESSOR_AGGREGATE)
+            )
         }
     }
 
@@ -145,4 +164,4 @@ open class MessagePrioritizeProcessor : AbstractMessagePrioritizeProcessor<ByteA
     open fun getGroupCorrelationTypes(messagePrioritization: MessagePrioritization): List<String>? {
         return null
     }
-}
\ No newline at end of file
+}
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageCorrelationUtils.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageCorrelationUtils.kt
index cc30af2f1..fb35df75b 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageCorrelationUtils.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageCorrelationUtils.kt
@@ -43,8 +43,8 @@ object MessageCorrelationUtils {
     }
 
     /** Assumption is message is of same group and checking for required types **/
-    fun correlatedMessagesWithTypes(collectedMessages: List<MessagePrioritization>, types: List<String>?)
-            : CorrelationCheckResponse {
+    fun correlatedMessagesWithTypes(collectedMessages: List<MessagePrioritization>, types: List<String>?):
+            CorrelationCheckResponse {
 
         return if (!types.isNullOrEmpty() && collectedMessages.size > 1) {
 
@@ -56,8 +56,8 @@ object MessageCorrelationUtils {
             val copyTypes = types.toTypedArray().copyOf().toMutableList()
 
             val filteredMessage = collectedMessages.filter {
-                !it.correlationId.isNullOrBlank()
-                        && types.contains(it.type)
+                !it.correlationId.isNullOrBlank() &&
+                        types.contains(it.type)
             }
             var correlatedKeys: MutableSet<String> = mutableSetOf()
             if (filteredMessage.isNotEmpty()) {
@@ -79,4 +79,4 @@ object MessageCorrelationUtils {
             return correlatedMessages(collectedMessages)
         }
     }
-}
\ No newline at end of file
+}
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessagePrioritizationSample.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessagePrioritizationSample.kt
index 185022973..4a36a40f3 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessagePrioritizationSample.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessagePrioritizationSample.kt
@@ -21,7 +21,9 @@ import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.E
 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.PrioritizationConfiguration
 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.ShutDownConfiguration
 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization
-import java.util.*
+import java.util.Calendar
+import java.util.Date
+import java.util.UUID
 
 object MessagePrioritizationSample {
 
@@ -57,8 +59,10 @@ object MessagePrioritizationSample {
     fun sampleMessages(groupName: String, messageState: String, count: Int): List<MessagePrioritization> {
         val messages: MutableList<MessagePrioritization> = arrayListOf()
         repeat(count) {
-            val backPressureMessage = createMessage(groupName, messageState,
-                    "sample-type", null)
+            val backPressureMessage = createMessage(
+                groupName, messageState,
+                "sample-type", null
+            )
             messages.add(backPressureMessage)
         }
         return messages
@@ -67,26 +71,37 @@ object MessagePrioritizationSample {
     fun sampleMessageWithSameCorrelation(groupName: String, messageState: String, count: Int): List<MessagePrioritization> {
         val messages: MutableList<MessagePrioritization> = arrayListOf()
         repeat(count) {
-            val backPressureMessage = createMessage(groupName, messageState, "sample-type",
-                    "key1=value1,key2=value2")
+            val backPressureMessage = createMessage(
+                groupName, messageState, "sample-type",
+                "key1=value1,key2=value2"
+            )
             messages.add(backPressureMessage)
         }
         return messages
     }
 
-    fun sampleMessageWithDifferentTypeSameCorrelation(groupName: String, messageState: String,
-                                                      count: Int): List<MessagePrioritization> {
+    fun sampleMessageWithDifferentTypeSameCorrelation(
+        groupName: String,
+        messageState: String,
+        count: Int
+    ): List<MessagePrioritization> {
         val messages: MutableList<MessagePrioritization> = arrayListOf()
         repeat(count) {
-            val backPressureMessage = createMessage(groupName, messageState, "type-$it",
-                    "key1=value1,key2=value2")
+            val backPressureMessage = createMessage(
+                groupName, messageState, "type-$it",
+                "key1=value1,key2=value2"
+            )
             messages.add(backPressureMessage)
         }
         return messages
     }
 
-    fun createMessage(groupName: String, messageState: String, messageType: String,
-                      messageCorrelationId: String?): MessagePrioritization {
+    fun createMessage(
+        groupName: String,
+        messageState: String,
+        messageType: String,
+        messageCorrelationId: String?
+    ): MessagePrioritization {
 
         return MessagePrioritization().apply {
             id = UUID.randomUUID().toString()
@@ -101,4 +116,4 @@ object MessagePrioritizationSample {
             expiryDate = currentDatePlusDays(3)
         }
     }
-}
\ No newline at end of file
+}
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 02614d821..7e5862cce 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
@@ -23,8 +23,8 @@ import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyS
 
 object MessageProcessorUtils {
 
-    fun <K, V> bluePrintProcessorSupplier(name: String, prioritizationConfiguration: PrioritizationConfiguration)
-            : ProcessorSupplier<K, V> {
+    fun <K, V> bluePrintProcessorSupplier(name: String, prioritizationConfiguration: PrioritizationConfiguration):
+            ProcessorSupplier<K, V> {
         return ProcessorSupplier<K, V> {
             // Dynamically resolve the Prioritization Processor
             val processorInstance = BluePrintDependencyService.instance<AbstractMessagePrioritizeProcessor<K, V>>(name)
@@ -32,5 +32,4 @@ object MessageProcessorUtils {
             processorInstance
         }
     }
-
-}
\ No newline at end of file
+}
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationConsumerTest.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationConsumerTest.kt
index 84f13dc1d..0ed9598f0 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationConsumerTest.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationConsumerTest.kt
@@ -44,29 +44,32 @@ import org.springframework.test.context.junit4.SpringRunner
 import kotlin.test.Test
 import kotlin.test.assertNotNull
 
-
 @RunWith(SpringRunner::class)
 @DataJpaTest
 @DirtiesContext
-@ContextConfiguration(classes = [BluePrintMessageLibConfiguration::class,
-    BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class,
-    MessagePrioritizationConfiguration::class, TestDatabaseConfiguration::class])
-@TestPropertySource(properties =
-[
-    "spring.jpa.show-sql=true",
-    "spring.jpa.properties.hibernate.show_sql=true",
-    "spring.jpa.hibernate.naming.physical-strategy=org.hibernate.boot.model.naming.PhysicalNamingStrategyStandardImpl",
-
-    "blueprintsprocessor.messageconsumer.prioritize-input.type=kafka-streams-basic-auth",
-    "blueprintsprocessor.messageconsumer.prioritize-input.bootstrapServers=127.0.0.1:9092",
-    "blueprintsprocessor.messageconsumer.prioritize-input.applicationId=test-prioritize-application",
-    "blueprintsprocessor.messageconsumer.prioritize-input.topic=prioritize-input-topic",
-
-    // To send initial test message
-    "blueprintsprocessor.messageproducer.prioritize-input.type=kafka-basic-auth",
-    "blueprintsprocessor.messageproducer.prioritize-input.bootstrapServers=127.0.0.1:9092",
-    "blueprintsprocessor.messageproducer.prioritize-input.topic=prioritize-input-topic"
-])
+@ContextConfiguration(
+    classes = [BluePrintMessageLibConfiguration::class,
+        BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class,
+        MessagePrioritizationConfiguration::class, TestDatabaseConfiguration::class]
+)
+@TestPropertySource(
+    properties =
+    [
+        "spring.jpa.show-sql=true",
+        "spring.jpa.properties.hibernate.show_sql=true",
+        "spring.jpa.hibernate.naming.physical-strategy=org.hibernate.boot.model.naming.PhysicalNamingStrategyStandardImpl",
+
+        "blueprintsprocessor.messageconsumer.prioritize-input.type=kafka-streams-basic-auth",
+        "blueprintsprocessor.messageconsumer.prioritize-input.bootstrapServers=127.0.0.1:9092",
+        "blueprintsprocessor.messageconsumer.prioritize-input.applicationId=test-prioritize-application",
+        "blueprintsprocessor.messageconsumer.prioritize-input.topic=prioritize-input-topic",
+
+        // To send initial test message
+        "blueprintsprocessor.messageproducer.prioritize-input.type=kafka-basic-auth",
+        "blueprintsprocessor.messageproducer.prioritize-input.bootstrapServers=127.0.0.1:9092",
+        "blueprintsprocessor.messageproducer.prioritize-input.topic=prioritize-input-topic"
+    ]
+)
 open class MessagePrioritizationConsumerTest {
 
     @Autowired
@@ -89,7 +92,7 @@ open class MessagePrioritizationConsumerTest {
             assertNotNull(prioritizationMessageRepository, "failed to get prioritizationMessageRepository")
 
             val messagePrioritizationService: MessagePrioritizationStateService = BluePrintDependencyService
-                    .instance(MessagePrioritizationStateService::class)
+                .instance(MessagePrioritizationStateService::class)
             assertNotNull(messagePrioritizationService, "failed to get messagePrioritizationService")
 
             MessagePrioritizationSample.sampleMessages(MessageState.NEW.name, 1).forEach {
@@ -106,7 +109,7 @@ open class MessagePrioritizationConsumerTest {
             val configuration = MessagePrioritizationSample.samplePrioritizationConfiguration()
 
             val streamingConsumerService = bluePrintMessageLibPropertyService
-                    .blueprintMessageConsumerService(configuration.inputTopicSelector)
+                .blueprintMessageConsumerService(configuration.inputTopicSelector)
             assertNotNull(streamingConsumerService, "failed to get blueprintMessageConsumerService")
 
             val spyStreamingConsumerService = spyk(streamingConsumerService)
@@ -115,11 +118,10 @@ open class MessagePrioritizationConsumerTest {
             val messagePrioritizationConsumer = MessagePrioritizationConsumer(bluePrintMessageLibPropertyService)
             val spyMessagePrioritizationConsumer = spyk(messagePrioritizationConsumer)
 
-
             // Test Topology
             val kafkaStreamConsumerFunction = spyMessagePrioritizationConsumer.kafkaStreamConsumerFunction(configuration)
             val messageConsumerProperties = bluePrintMessageLibPropertyService
-                    .messageConsumerProperties("blueprintsprocessor.messageconsumer.prioritize-input")
+                .messageConsumerProperties("blueprintsprocessor.messageconsumer.prioritize-input")
             val topology = kafkaStreamConsumerFunction.createTopology(messageConsumerProperties, null)
             assertNotNull(topology, "failed to get create topology")
 
@@ -130,7 +132,7 @@ open class MessagePrioritizationConsumerTest {
     }
 
     /** Integration Kafka Testing, Enable and use this test case only for local desktop testing with real kafka broker */
-    //@Test
+    // @Test
     fun testMessagePrioritizationConsumer() {
         runBlocking {
             val messagePrioritizationConsumer = MessagePrioritizationConsumer(bluePrintMessageLibPropertyService)
@@ -138,38 +140,44 @@ open class MessagePrioritizationConsumerTest {
 
             /** Send sample message with every 1 sec */
             val blueprintMessageProducerService = bluePrintMessageLibPropertyService
-                    .blueprintMessageProducerService("prioritize-input") as KafkaBasicAuthMessageProducerService
+                .blueprintMessageProducerService("prioritize-input") as KafkaBasicAuthMessageProducerService
             launch {
-             MessagePrioritizationSample.sampleMessages(MessageState.NEW.name, 2).forEach {
+                MessagePrioritizationSample.sampleMessages(MessageState.NEW.name, 2).forEach {
                     delay(100)
                     val headers: MutableMap<String, String> = hashMapOf()
                     headers["id"] = it.id
-                    blueprintMessageProducerService.sendMessageNB(message = it.asJsonString(false),
-                            headers = headers)
+                    blueprintMessageProducerService.sendMessageNB(
+                        message = it.asJsonString(false),
+                        headers = headers
+                    )
                 }
 
                 MessagePrioritizationSample
-                        .sampleMessageWithSameCorrelation("same-group", MessageState.NEW.name, 2)
-                        .forEach {
-                            delay(100)
-                            val headers: MutableMap<String, String> = hashMapOf()
-                            headers["id"] = it.id
-                            blueprintMessageProducerService.sendMessageNB(message = it.asJsonString(false),
-                                    headers = headers)
-                        }
+                    .sampleMessageWithSameCorrelation("same-group", MessageState.NEW.name, 2)
+                    .forEach {
+                        delay(100)
+                        val headers: MutableMap<String, String> = hashMapOf()
+                        headers["id"] = it.id
+                        blueprintMessageProducerService.sendMessageNB(
+                            message = it.asJsonString(false),
+                            headers = headers
+                        )
+                    }
 
                 MessagePrioritizationSample
-                        .sampleMessageWithDifferentTypeSameCorrelation("group-typed", MessageState.NEW.name, 3)
-                        .forEach {
-                            delay(2000)
-                            val headers: MutableMap<String, String> = hashMapOf()
-                            headers["id"] = it.id
-                            blueprintMessageProducerService.sendMessageNB(message = it.asJsonString(false),
-                                    headers = headers)
-                        }
+                    .sampleMessageWithDifferentTypeSameCorrelation("group-typed", MessageState.NEW.name, 3)
+                    .forEach {
+                        delay(2000)
+                        val headers: MutableMap<String, String> = hashMapOf()
+                        headers["id"] = it.id
+                        blueprintMessageProducerService.sendMessageNB(
+                            message = it.asJsonString(false),
+                            headers = headers
+                        )
+                    }
             }
             delay(10000)
             messagePrioritizationConsumer.shutDown()
         }
     }
-}
\ No newline at end of file
+}
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/TestConfiguration.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/TestConfiguration.kt
index 4e3eb191b..be65c1d7c 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/TestConfiguration.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/TestConfiguration.kt
@@ -42,6 +42,7 @@ open class TestDatabaseConfiguration {
 
 @Service(MessagePrioritizationConstants.PROCESSOR_PRIORITIZE)
 open class TestMessagePrioritizeProcessor : MessagePrioritizeProcessor() {
+
     override fun getGroupCorrelationTypes(messagePrioritization: MessagePrioritization): List<String>? {
         return when (messagePrioritization.group) {
             "group-typed" -> arrayListOf("type-0", "type-1", "type-2")
@@ -54,4 +55,4 @@ open class TestMessagePrioritizeProcessor : MessagePrioritizeProcessor() {
 open class DefaultMessageAggregateProcessor() : MessageAggregateProcessor()
 
 @Service(MessagePrioritizationConstants.PROCESSOR_OUTPUT)
-open class DefaultMessageOutputProcessor : MessageOutputProcessor()
\ No newline at end of file
+open class DefaultMessageOutputProcessor : MessageOutputProcessor()
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageCorrelationUtilsTest.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageCorrelationUtilsTest.kt
index b470db909..3876cbba5 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageCorrelationUtilsTest.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageCorrelationUtilsTest.kt
@@ -26,10 +26,14 @@ class MessageCorrelationUtilsTest {
     @Test
     fun testCorrelationKeysReordered() {
 
-        val message1 = MessagePrioritizationSample.createMessage("sample-group", MessageState.NEW.name,
-                "type-0", "key1=value1,key2=value2")
-        val message2 = MessagePrioritizationSample.createMessage("sample-group", MessageState.NEW.name,
-                "type-0", "key2=value2,key1=value1")
+        val message1 = MessagePrioritizationSample.createMessage(
+            "sample-group", MessageState.NEW.name,
+            "type-0", "key1=value1,key2=value2"
+        )
+        val message2 = MessagePrioritizationSample.createMessage(
+            "sample-group", MessageState.NEW.name,
+            "type-0", "key2=value2,key1=value1"
+        )
 
         val multipleMessages: MutableList<MessagePrioritization> = arrayListOf()
         multipleMessages.add(message1)
@@ -43,20 +47,26 @@ class MessageCorrelationUtilsTest {
         /** With Types **/
         /* Assumption is Same group with different types */
         val differentTypesWithSameCorrelationMessages = MessagePrioritizationSample
-                .sampleMessageWithDifferentTypeSameCorrelation("sample-group", MessageState.NEW.name, 3)
+            .sampleMessageWithDifferentTypeSameCorrelation("sample-group", MessageState.NEW.name, 3)
         val differentTypesWithSameCorrelationMessagesResponse = MessageCorrelationUtils.correlatedMessagesWithTypes(
-                differentTypesWithSameCorrelationMessages,
-                arrayListOf("type-0", "type-1", "type-2"))
-        assertTrue(differentTypesWithSameCorrelationMessagesResponse.correlated,
-                "failed to correlate differentTypesWithSameCorrelationMessagesResponse")
+            differentTypesWithSameCorrelationMessages,
+            arrayListOf("type-0", "type-1", "type-2")
+        )
+        assertTrue(
+            differentTypesWithSameCorrelationMessagesResponse.correlated,
+            "failed to correlate differentTypesWithSameCorrelationMessagesResponse"
+        )
 
         /* Assumption is Same group with different types and one missing expected types,
         In this case type-3 message is missing */
         val differentTypesWithSameCorrelationMessagesResWithMissingType = MessageCorrelationUtils.correlatedMessagesWithTypes(
-                differentTypesWithSameCorrelationMessages,
-                arrayListOf("type-0", "type-1", "type-2", "type-3"))
-        assertTrue(!differentTypesWithSameCorrelationMessagesResWithMissingType.correlated,
-                "failed to correlate differentTypesWithSameCorrelationMessagesResWithMissingType")
+            differentTypesWithSameCorrelationMessages,
+            arrayListOf("type-0", "type-1", "type-2", "type-3")
+        )
+        assertTrue(
+            !differentTypesWithSameCorrelationMessagesResWithMissingType.correlated,
+            "failed to correlate differentTypesWithSameCorrelationMessagesResWithMissingType"
+        )
     }
 
     @Test
@@ -64,35 +74,48 @@ class MessageCorrelationUtilsTest {
         /** With ignoring Types */
         /** Assumption is only one message received */
         val withSameCorrelationOneMessages = MessagePrioritizationSample
-                .sampleMessageWithSameCorrelation("sample-group", MessageState.NEW.name, 1)
+            .sampleMessageWithSameCorrelation("sample-group", MessageState.NEW.name, 1)
         val withSameCorrelationOneMessagesResp = MessageCorrelationUtils.correlatedMessagesWithTypes(
-                withSameCorrelationOneMessages, null)
-        assertTrue(!withSameCorrelationOneMessagesResp.correlated,
-                "failed to correlate withSameCorrelationMessagesResp")
+            withSameCorrelationOneMessages, null
+        )
+        assertTrue(
+            !withSameCorrelationOneMessagesResp.correlated,
+            "failed to correlate withSameCorrelationMessagesResp"
+        )
 
         /** Assumption is two message received for same group with same correlation */
         val withSameCorrelationMessages = MessagePrioritizationSample
-                .sampleMessageWithSameCorrelation("sample-group", MessageState.NEW.name, 2)
+            .sampleMessageWithSameCorrelation("sample-group", MessageState.NEW.name, 2)
         val withSameCorrelationMessagesResp = MessageCorrelationUtils.correlatedMessagesWithTypes(
-                withSameCorrelationMessages, null)
-        assertTrue(withSameCorrelationMessagesResp.correlated,
-                "failed to correlate withSameCorrelationMessagesResp")
+            withSameCorrelationMessages, null
+        )
+        assertTrue(
+            withSameCorrelationMessagesResp.correlated,
+            "failed to correlate withSameCorrelationMessagesResp"
+        )
     }
 
     @Test
     fun differentTypesWithDifferentCorrelationMessage() {
         /** Assumption is two message received for same group with different expected types and different correlation */
-        val message1 = MessagePrioritizationSample.createMessage("sample-group", MessageState.NEW.name,
-                "type-0", "key1=value1,key2=value2")
-        val message2 = MessagePrioritizationSample.createMessage("sample-group", MessageState.NEW.name,
-                "type-1", "key1=value1,key2=value3")
+        val message1 = MessagePrioritizationSample.createMessage(
+            "sample-group", MessageState.NEW.name,
+            "type-0", "key1=value1,key2=value2"
+        )
+        val message2 = MessagePrioritizationSample.createMessage(
+            "sample-group", MessageState.NEW.name,
+            "type-1", "key1=value1,key2=value3"
+        )
         val differentTypesWithDifferentCorrelationMessage: MutableList<MessagePrioritization> = arrayListOf()
         differentTypesWithDifferentCorrelationMessage.add(message1)
         differentTypesWithDifferentCorrelationMessage.add(message2)
         val differentTypesWithDifferentCorrelationMessageResp = MessageCorrelationUtils.correlatedMessagesWithTypes(
-                differentTypesWithDifferentCorrelationMessage,
-                arrayListOf("type-0", "type-1"))
-        assertTrue(!differentTypesWithDifferentCorrelationMessageResp.correlated,
-                "failed to correlate differentTypesWithDifferentCorrelationMessageResp")
+            differentTypesWithDifferentCorrelationMessage,
+            arrayListOf("type-0", "type-1")
+        )
+        assertTrue(
+            !differentTypesWithDifferentCorrelationMessageResp.correlated,
+            "failed to correlate differentTypesWithDifferentCorrelationMessageResp"
+        )
     }
-}
\ No newline at end of file
+}