diff options
Diffstat (limited to 'ms/blueprintsprocessor')
20 files changed, 448 insertions, 352 deletions
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/README.md b/ms/blueprintsprocessor/functions/message-prioritizaion/README.md index 482bbc2cc..cda43faca 100644 --- a/ms/blueprintsprocessor/functions/message-prioritizaion/README.md +++ b/ms/blueprintsprocessor/functions/message-prioritizaion/README.md @@ -3,7 +3,6 @@ To Delete Topics ------------------ kafka-topics --zookeeper localhost:2181 --delete --topic prioritize-input-topic kafka-topics --zookeeper localhost:2181 --delete --topic prioritize-output-topic -kafka-topics --zookeeper localhost:2181 --delete --topic prioritize-expired-topic kafka-topics --zookeeper localhost:2181 --delete --topic test-prioritize-application-PriorityMessage-changelog Create Topics @@ -11,7 +10,6 @@ Create Topics kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic prioritize-input-topic kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic prioritize-output-topic -kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic prioritize-expired-topic To List topics ---------------- @@ -26,6 +24,3 @@ To Listen for Output kafka-console-consumer --bootstrap-server localhost:9092 --topic prioritize-output-topic --from-beginning kafka-console-consumer --bootstrap-server localhost:9092 --topic prioritize-input-topic --from-beginning - -kafka-console-consumer --bootstrap-server localhost:9092 --topic prioritize-expired-topic --from-beginning - 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 28e096352..890e0a6ba 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 @@ -28,9 +28,6 @@ object MessagePrioritizationConstants { const val SOURCE_INPUT = "source-prioritization-input" const val PROCESSOR_PRIORITIZE = "processor-prioritization-prioritize" - const val PROCESSOR_AGGREGATE = "processor-prioritization-aggregate" - const val PROCESSOR_OUTPUT = "processor-prioritization-output" const val SINK_OUTPUT = "sink-prioritization-output" - const val SINK_EXPIRED = "sink-prioritization-expired" } diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationService.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationService.kt new file mode 100644 index 000000000..584fd00d3 --- /dev/null +++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationService.kt @@ -0,0 +1,29 @@ +/* + * 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 + +import org.apache.kafka.streams.processor.ProcessorContext +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization + +interface MessagePrioritizationService { + + fun setKafkaProcessorContext(processorContext: ProcessorContext?) + + suspend fun prioritize(messagePrioritization: MessagePrioritization) + + suspend fun output(id: String) +} diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationStateService.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationStateService.kt new file mode 100644 index 000000000..5dd41d7f3 --- /dev/null +++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationStateService.kt @@ -0,0 +1,68 @@ +/* + * 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 + +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization +import java.util.Date + +interface MessagePrioritizationStateService { + + suspend fun saveMessage(message: MessagePrioritization): MessagePrioritization + + suspend fun getMessage(id: String): MessagePrioritization + + suspend fun getMessages(ids: List<String>): List<MessagePrioritization>? + + suspend fun getExpiryEligibleMessages(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 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 updateMessagesState(ids: List<String>, state: String) + + suspend fun updateMessageState(id: String, state: String): MessagePrioritization + + suspend fun setMessageState(id: String, state: String) + + suspend fun setMessagesPriority(ids: List<String>, priority: String) + + suspend fun setMessagesState(ids: List<String>, state: String) + + suspend fun setMessageStateANdError(id: String, state: String, error: String) + + suspend fun setMessageStateAndAggregatedIds(id: String, state: String, aggregatedIds: List<String>) + + suspend fun deleteMessage(id: String) + + suspend fun deleteMessageByGroup(group: String) + + suspend fun deleteMessageStates(group: String, states: List<String>) + + suspend fun deleteExpiredMessage(group: String, retentionDays: Int) +} 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 bef7a7b61..05b820adb 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 @@ -17,7 +17,6 @@ package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization -import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.service.MessagePrioritizationStateService import org.onap.ccsdk.cds.blueprintsprocessor.services.execution.AbstractComponentFunction import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService 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 262dcb402..e90771fb8 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 @@ -16,9 +16,10 @@ package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.api +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationService +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationStateService import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.UpdateStateRequest import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization -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.GetMapping @@ -31,7 +32,10 @@ import org.springframework.web.bind.annotation.RestController @RestController @RequestMapping(value = ["/api/v1/message-prioritization"]) -open class MessagePrioritizationApi(private val messagePrioritizationStateService: MessagePrioritizationStateService) { +open class MessagePrioritizationApi( + private val messagePrioritizationStateService: MessagePrioritizationStateService, + private val messagePrioritizationService: MessagePrioritizationService +) { @GetMapping(path = ["/ping"], produces = [MediaType.APPLICATION_JSON_VALUE]) @ResponseBody @@ -53,6 +57,15 @@ open class MessagePrioritizationApi(private val messagePrioritizationStateServic } @PostMapping( + path = ["/prioritize"], produces = [MediaType.APPLICATION_JSON_VALUE], + consumes = [MediaType.APPLICATION_JSON_VALUE] + ) + @ResponseBody + fun prioritize(@RequestBody messagePrioritization: MessagePrioritization) = monoMdc { + messagePrioritizationService.prioritize(messagePrioritization) + } + + @PostMapping( path = ["/update-state"], produces = [MediaType.APPLICATION_JSON_VALUE], consumes = [MediaType.APPLICATION_JSON_VALUE] ) diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/AbstractMessagePrioritizeProcessor.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/kafka/AbstractMessagePrioritizeProcessor.kt index 35566abb4..656646ff7 100644 --- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/AbstractMessagePrioritizeProcessor.kt +++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/kafka/AbstractMessagePrioritizeProcessor.kt @@ -14,38 +14,21 @@ * limitations under the License. */ -package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization +package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.kafka import org.apache.kafka.streams.processor.ProcessorContext -import org.onap.ccsdk.cds.blueprintsprocessor.atomix.clusterService -import org.onap.ccsdk.cds.blueprintsprocessor.core.service.BluePrintClusterService -import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.service.MessagePrioritizationStateService +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.PrioritizationConfiguration import org.onap.ccsdk.cds.blueprintsprocessor.message.kafka.AbstractBluePrintMessageProcessor -import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants import org.onap.ccsdk.cds.controllerblueprints.core.logger -import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService -/** CDS Message Prioritazation Kafka Stream Processor abstract class to implement */ +/** CDS Message Prioritization Kafka Stream Processor abstract class to implement */ abstract class AbstractMessagePrioritizeProcessor<K, V> : AbstractBluePrintMessageProcessor<K, V>() { private val log = logger(AbstractMessagePrioritizeProcessor::class) lateinit var prioritizationConfiguration: PrioritizationConfiguration - lateinit var messagePrioritizationStateService: MessagePrioritizationStateService - var clusterService: BluePrintClusterService? = null - override fun init(context: ProcessorContext) { - this.processorContext = context - /** Get the State service to update in store */ - this.messagePrioritizationStateService = BluePrintDependencyService - .messagePrioritizationStateService() - } - - /** Cluster Service is not enabled by default for all processors, In needed initialize from processor init method */ - open fun initializeClusterService() { - /** Get the Cluster service to update in store */ - if (BluePrintConstants.CLUSTER_ENABLED) { - this.clusterService = BluePrintDependencyService.clusterService() - } + override fun init(processorContext: ProcessorContext) { + this.processorContext = processorContext } } diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/kafka/DefaultMessagePrioritizeProcessor.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/kafka/DefaultMessagePrioritizeProcessor.kt new file mode 100644 index 000000000..c14a404ad --- /dev/null +++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/kafka/DefaultMessagePrioritizeProcessor.kt @@ -0,0 +1,115 @@ +/* + * 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.kafka + +import org.apache.kafka.streams.processor.Cancellable +import org.apache.kafka.streams.processor.ProcessorContext +import org.apache.kafka.streams.processor.PunctuationType +import org.apache.kafka.streams.processor.To +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationConstants +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationService +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationStateService +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessageState +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization +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 + +open class DefaultMessagePrioritizeProcessor( + private val messagePrioritizationStateService: MessagePrioritizationStateService, + private val messagePrioritizationService: MessagePrioritizationService +) : AbstractMessagePrioritizeProcessor<ByteArray, ByteArray>() { + + private val log = logger(DefaultMessagePrioritizeProcessor::class) + + lateinit var expiryCancellable: Cancellable + lateinit var cleanCancellable: Cancellable + + override suspend fun processNB(key: ByteArray, value: ByteArray) { + + val messagePrioritize = JacksonUtils.readValue(String(value), MessagePrioritization::class.java) + ?: throw BluePrintProcessorException("failed to convert") + try { + messagePrioritizationService.setKafkaProcessorContext(processorContext) + messagePrioritizationService.prioritize(messagePrioritize) + } catch (e: Exception) { + 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!! + ) + /** Publish to Output topic */ + this.processorContext.forward( + messagePrioritize.id, messagePrioritize, + To.child(MessagePrioritizationConstants.SINK_OUTPUT) + ) + } + } + + override fun init(context: ProcessorContext) { + super.init(context) + /** set up expiry marking cron */ + initializeExpiryPunctuator() + /** Set up cleaning records cron */ + initializeCleanPunctuator() + } + + override fun close() { + log.info( + "closing prioritization processor applicationId(${processorContext.applicationId()}), " + + "taskId(${processorContext.taskId()})" + ) + expiryCancellable.cancel() + cleanCancellable.cancel() + } + + open fun initializeExpiryPunctuator() { + val expiryPunctuator = + MessagePriorityExpiryPunctuator( + messagePrioritizationStateService + ) + expiryPunctuator.processorContext = processorContext + expiryPunctuator.configuration = prioritizationConfiguration + val expiryConfiguration = prioritizationConfiguration.expiryConfiguration + expiryCancellable = processorContext.schedule( + Duration.ofMillis(expiryConfiguration.frequencyMilli), + PunctuationType.WALL_CLOCK_TIME, expiryPunctuator + ) + log.info("Expiry punctuator setup complete with frequency(${expiryConfiguration.frequencyMilli})mSec") + } + + open fun initializeCleanPunctuator() { + val cleanPunctuator = + MessagePriorityCleanPunctuator( + messagePrioritizationStateService + ) + 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" + ) + } +} 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/kafka/MessagePrioritizationConsumer.kt index b611060f7..d7666a20b 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/kafka/MessagePrioritizationConsumer.kt @@ -14,11 +14,12 @@ * limitations under the License. */ -package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization +package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.kafka import org.apache.kafka.common.serialization.Serdes import org.apache.kafka.streams.Topology -import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.topology.MessagePrioritizationSerde +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationConstants +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.PrioritizationConfiguration import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.utils.MessageProcessorUtils.bluePrintProcessorSupplier import org.onap.ccsdk.cds.blueprintsprocessor.message.KafkaStreamsBasicAuthConsumerProperties import org.onap.ccsdk.cds.blueprintsprocessor.message.MessageConsumerProperties @@ -42,7 +43,7 @@ open class MessagePrioritizationConsumer( } open fun kafkaStreamConsumerFunction(prioritizationConfiguration: PrioritizationConfiguration): - KafkaStreamConsumerFunction { + KafkaStreamConsumerFunction { return object : KafkaStreamConsumerFunction { override suspend fun createTopology( @@ -52,7 +53,7 @@ open class MessagePrioritizationConsumer( val topology = Topology() val kafkaStreamsBasicAuthConsumerProperties = messageConsumerProperties - as KafkaStreamsBasicAuthConsumerProperties + as KafkaStreamsBasicAuthConsumerProperties val topics = kafkaStreamsBasicAuthConsumerProperties.topic.splitCommaAsList() log.info("Consuming prioritization topics($topics)") @@ -68,39 +69,12 @@ open class MessagePrioritizationConsumer( 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(), - MessagePrioritizationConstants.PROCESSOR_PRIORITIZE, - MessagePrioritizationConstants.PROCESSOR_AGGREGATE, - MessagePrioritizationConstants.PROCESSOR_OUTPUT + MessagePrioritizationConstants.PROCESSOR_PRIORITIZE ) // Output will be sent to the group-output topic from Processor API 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/kafka/MessagePrioritizationPunctuators.kt index 5435ebe30..e27cf16d0 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/kafka/MessagePrioritizationPunctuators.kt @@ -14,13 +14,13 @@ * limitations under the License. */ -package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.topology +package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.kafka import org.apache.kafka.streams.processor.To import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationConstants +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationStateService import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessageState import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.PrioritizationConfiguration -import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.service.MessagePrioritizationStateService import org.onap.ccsdk.cds.blueprintsprocessor.message.kafka.AbstractBluePrintMessagePunctuator import org.onap.ccsdk.cds.controllerblueprints.core.logger @@ -46,7 +46,7 @@ class MessagePriorityExpiryPunctuator(private val messagePrioritizationStateServ fetchMessages.forEach { expired -> processorContext.forward( expired.id, expired, - To.child(MessagePrioritizationConstants.SINK_EXPIRED) + To.child(MessagePrioritizationConstants.SINK_OUTPUT) ) } } diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessagePrioritizationSerde.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/kafka/MessagePrioritizationSerde.kt index f2a481f74..5595863d4 100644 --- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessagePrioritizationSerde.kt +++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/kafka/MessagePrioritizationSerde.kt @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.topology +package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.kafka import org.apache.kafka.common.serialization.Deserializer import org.apache.kafka.common.serialization.Serde 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/service/AbstractMessagePrioritizationService.kt index 4e4e2da7a..13c0dd7bc 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/service/AbstractMessagePrioritizationService.kt @@ -14,43 +14,42 @@ * limitations under the License. */ -package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.topology +package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.service -import org.apache.kafka.streams.processor.Cancellable import org.apache.kafka.streams.processor.ProcessorContext -import org.apache.kafka.streams.processor.PunctuationType import org.apache.kafka.streams.processor.To -import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.AbstractMessagePrioritizeProcessor import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationConstants +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationService +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationStateService import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessageState import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.utils.MessageCorrelationUtils import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.utils.MessageProcessorUtils -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.UUID -open class MessagePrioritizeProcessor : AbstractMessagePrioritizeProcessor<ByteArray, ByteArray>() { +/** Child should implement with sequencing & aggregation handling along with group type correlation mappings.*/ +abstract class AbstractMessagePrioritizationService( + private val messagePrioritizationStateService: MessagePrioritizationStateService +) : MessagePrioritizationService { - private val log = logger(MessagePrioritizeProcessor::class) + private val log = logger(AbstractMessagePrioritizationService::class) - lateinit var expiryCancellable: Cancellable - lateinit var cleanCancellable: Cancellable + var processorContext: ProcessorContext? = null - 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") + override fun setKafkaProcessorContext(processorContext: ProcessorContext?) { + this.processorContext = processorContext + } + + override suspend fun prioritize(messagePrioritize: MessagePrioritization) { try { + log.info("***** received in prioritize processor key(${messagePrioritize.id})") /** Get the cluster lock for message group */ - val clusterLock = MessageProcessorUtils.prioritizationGrouplock(clusterService, messagePrioritize) + val clusterLock = MessageProcessorUtils.prioritizationGrouplock(messagePrioritize) // Save the Message messagePrioritizationStateService.saveMessage(messagePrioritize) handleCorrelationAndNextStep(messagePrioritize) /** Cluster unLock for message group */ - MessageProcessorUtils.prioritizationGroupUnLock(clusterService, clusterLock) + MessageProcessorUtils.prioritizationGroupUnLock(clusterLock) } catch (e: Exception) { messagePrioritize.error = "failed in Prioritize message(${messagePrioritize.id}) : ${e.message}" log.error(messagePrioritize.error) @@ -59,58 +58,16 @@ open class MessagePrioritizeProcessor : AbstractMessagePrioritizeProcessor<ByteA messagePrioritize.id, MessageState.ERROR.name, messagePrioritize.error!! ) - /** Publish to Output topic */ - this.processorContext.forward( - messagePrioritize.id, messagePrioritize, - To.child(MessagePrioritizationConstants.SINK_OUTPUT) - ) } } - override fun init(context: ProcessorContext) { - super.init(context) - /** set up expiry marking cron */ - initializeExpiryPunctuator() - /** Set up cleaning records cron */ - initializeCleanPunctuator() - /** Set up Cluster Service */ - initializeClusterService() - } - - override fun close() { - log.info( - "closing prioritization processor applicationId(${processorContext.applicationId()}), " + - "taskId(${processorContext.taskId()})" - ) - expiryCancellable.cancel() - cleanCancellable.cancel() - } - - open fun initializeExpiryPunctuator() { - val expiryPunctuator = MessagePriorityExpiryPunctuator(messagePrioritizationStateService) - expiryPunctuator.processorContext = processorContext - expiryPunctuator.configuration = prioritizationConfiguration - val expiryConfiguration = prioritizationConfiguration.expiryConfiguration - expiryCancellable = processorContext.schedule( - Duration.ofMillis(expiryConfiguration.frequencyMilli), - PunctuationType.WALL_CLOCK_TIME, expiryPunctuator - ) - log.info("Expiry punctuator setup complete with frequency(${expiryConfiguration.frequencyMilli})mSec") - } - - open fun initializeCleanPunctuator() { - val cleanPunctuator = MessagePriorityCleanPunctuator(messagePrioritizationStateService) - 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" - ) + override suspend fun output(id: String) { + log.info("$$$$$ received in output processor id($id)") + val message = messagePrioritizationStateService.updateMessageState(id, MessageState.COMPLETED.name) + /** Check for Kafka Processing, If yes, then send to the output topic */ + if (this.processorContext != null) { + processorContext!!.forward(message.id, message, To.child(MessagePrioritizationConstants.SINK_OUTPUT)) + } } open suspend fun handleCorrelationAndNextStep(messagePrioritization: MessagePrioritization) { @@ -126,10 +83,11 @@ open class MessagePrioritizeProcessor : AbstractMessagePrioritizeProcessor<ByteA ) /** 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) { @@ -139,12 +97,9 @@ open class MessagePrioritizeProcessor : AbstractMessagePrioritizeProcessor<ByteA 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) - ) + val correlatedIds = waitingCorrelatedStoreMessages.joinToString(",") { it.id } + /** Send only correlated ids to aggregate processor */ + aggregate(correlatedIds) } else { /** Correlation not satisfied */ log.trace("correlation not matched : ${correlationResults.message}") @@ -159,16 +114,57 @@ 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) - ) + aggregate(messagePrioritization.id) + } + } + + open suspend fun aggregate(strIds: String) { + log.info("@@@@@ received in aggregation processor ids($strIds)") + val ids = strIds.split(",").map { it.trim() } + if (!ids.isNullOrEmpty()) { + try { + if (ids.size == 1) { + /** No aggregation or sequencing needed, simpley forward to next processor */ + output(ids.first()) + } else { + /** Implement Aggregation logic in overridden class, If necessary, + Populate New Message and Update status with Prioritized, Forward the message to next processor */ + handleAggregation(ids) + /** Update all messages to Aggregated state */ + messagePrioritizationStateService.setMessagesState(ids, MessageState.AGGREGATED.name) + } + } catch (e: Exception) { + val error = "failed in Aggregate message($ids) : ${e.message}" + log.error(error, e) + val storeMessages = messagePrioritizationStateService.getMessages(ids) + if (!storeMessages.isNullOrEmpty()) { + storeMessages.forEach { messagePrioritization -> + try { + /** Update the data store */ + messagePrioritizationStateService.setMessageStateANdError( + messagePrioritization.id, + MessageState.ERROR.name, error + ) + /** Publish to output topic */ + output(messagePrioritization.id) + } catch (sendException: Exception) { + log.error( + "failed to update/publish error message(${messagePrioritization.id}) : " + + "${sendException.message}", e + ) + } + } + } + } } } + /** Child will override this implementation , if necessary + * Here the place child has to implement custom Sequencing and Aggregation logic. + * */ + abstract suspend fun handleAggregation(messageIds: List<String>) + /** If consumer wants specific correlation with respect to group and types, then populate the specific types, * otherwise correlation happens with group and correlationId */ - open fun getGroupCorrelationTypes(messagePrioritization: MessagePrioritization): List<String>? { - return null - } + abstract fun getGroupCorrelationTypes(messagePrioritization: MessagePrioritization): 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/MessagePrioritizationStateServiceImpl.kt index 017658ff6..d9cd956bf 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/MessagePrioritizationStateServiceImpl.kt @@ -16,6 +16,7 @@ package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.service +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationStateService import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessageState import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.PrioritizationMessageRepository @@ -27,59 +28,10 @@ import org.springframework.stereotype.Service import org.springframework.transaction.annotation.Transactional import java.util.Date -interface MessagePrioritizationStateService { - - suspend fun saveMessage(message: MessagePrioritization): MessagePrioritization - - suspend fun getMessage(id: String): MessagePrioritization - - suspend fun getMessages(ids: List<String>): List<MessagePrioritization>? - - suspend fun getExpiryEligibleMessages(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 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 updateMessagesState(ids: List<String>, state: String) - - suspend fun updateMessageState(id: String, state: String): MessagePrioritization - - suspend fun setMessageState(id: String, state: String) - - suspend fun setMessagesPriority(ids: List<String>, priority: String) - - suspend fun setMessagesState(ids: List<String>, state: String) - - suspend fun setMessageStateANdError(id: String, state: String, error: String) - - suspend fun setMessageStateAndAggregatedIds(id: String, state: String, aggregatedIds: List<String>) - - suspend fun deleteMessage(id: String) - - suspend fun deleteMessageByGroup(group: String) - - suspend fun deleteMessageStates(group: String, states: List<String>) - - suspend fun deleteExpiredMessage(group: String, retentionDays: Int) -} - @Service open class MessagePrioritizationStateServiceImpl( private val prioritizationMessageRepository: PrioritizationMessageRepository -) : - MessagePrioritizationStateService { +) : MessagePrioritizationStateService { private val log = logger(MessagePrioritizationStateServiceImpl::class) @@ -110,7 +62,7 @@ open class MessagePrioritizationStateServiceImpl( } override suspend fun getMessageForStatesNotExpiredIn(group: String, states: List<String>, count: Int): - List<MessagePrioritization>? { + List<MessagePrioritization>? { return prioritizationMessageRepository.findByGroupAndStateInAndNotExpiredDate( group, states, Date(), PageRequest.of(0, count) @@ -118,7 +70,7 @@ open class MessagePrioritizationStateServiceImpl( } override suspend fun getMessageForStatesExpired(group: String, states: List<String>, count: Int): - List<MessagePrioritization>? { + List<MessagePrioritization>? { return prioritizationMessageRepository.findByGroupAndStateInAndExpiredDate( group, states, Date(), PageRequest.of(0, count) @@ -126,7 +78,7 @@ open class MessagePrioritizationStateServiceImpl( } override suspend fun getExpiredMessages(group: String, expiryDate: Date, count: Int): - List<MessagePrioritization>? { + List<MessagePrioritization>? { return prioritizationMessageRepository.findByByGroupAndExpiredDate( group, expiryDate, PageRequest.of(0, count) diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/service/SampleMessagePrioritizationService.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/service/SampleMessagePrioritizationService.kt new file mode 100644 index 000000000..fcdb71cda --- /dev/null +++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/service/SampleMessagePrioritizationService.kt @@ -0,0 +1,46 @@ +/* + * 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.service + +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationStateService +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.kafka.DefaultMessagePrioritizeProcessor +import org.onap.ccsdk.cds.controllerblueprints.core.logger + +open class SampleMessagePrioritizationService(messagePrioritizationStateService: MessagePrioritizationStateService) : + AbstractMessagePrioritizationService(messagePrioritizationStateService) { + + private val log = logger(DefaultMessagePrioritizeProcessor::class) + + /** Child overriding this implementation , if necessary */ + override suspend fun handleAggregation(messageIds: List<String>) { + log.info("messages($messageIds) aggregated") + messageIds.forEach { id -> + output(id) + } + } + + /** If consumer wants specific correlation with respect to group and types, then populate the specific types, + * otherwise correlation happens with group and correlationId */ + override fun getGroupCorrelationTypes(messagePrioritization: MessagePrioritization): List<String>? { + return when (messagePrioritization.group) { + /** Dummy Implementation, This can also be read from file and stored as cached map **/ + "group-typed" -> arrayListOf("type-0", "type-1", "type-2") + else -> null + } + } +} 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 deleted file mode 100644 index 3e697e633..000000000 --- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessageAggregateProcessor.kt +++ /dev/null @@ -1,80 +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.topology - -import org.apache.kafka.streams.processor.To -import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.AbstractMessagePrioritizeProcessor -import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationConstants -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) - - override suspend fun processNB(key: String, value: String) { - - log.info("@@@@@ received in aggregation processor key($key), value($value)") - val ids = value.split(",").map { it.trim() } - if (!ids.isNullOrEmpty()) { - try { - if (ids.size == 1) { - processorContext.forward(key, ids.first(), To.child(MessagePrioritizationConstants.PROCESSOR_OUTPUT)) - } else { - /** Implement Aggregation logic in overridden class, If necessary, - Populate New Message and Update status with Prioritized, Forward the message to next processor */ - handleAggregation(ids) - /** Update all messages to Aggregated state */ - messagePrioritizationStateService.setMessagesState(ids, MessageState.AGGREGATED.name) - } - } catch (e: Exception) { - val error = "failed in Aggregate message($ids) : ${e.message}" - log.error(error, e) - val storeMessages = messagePrioritizationStateService.getMessages(ids) - if (!storeMessages.isNullOrEmpty()) { - storeMessages.forEach { messagePrioritization -> - try { - /** Update the data store */ - messagePrioritizationStateService.setMessageStateANdError( - messagePrioritization.id, - MessageState.ERROR.name, error - ) - /** Publish to Error topic */ - 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 - ) - } - } - } - } - } - } - - /** Child will override this implementation , if necessary */ - open suspend fun handleAggregation(messageIds: List<String>) { - log.info("messages($messageIds) aggregated") - messageIds.forEach { id -> - processorContext.forward(id, id, To.child(MessagePrioritizationConstants.PROCESSOR_OUTPUT)) - } - } -} 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 deleted file mode 100644 index cf6520df5..000000000 --- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/topology/MessageOutputProcessor.kt +++ /dev/null @@ -1,34 +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.topology - -import org.apache.kafka.streams.processor.To -import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.AbstractMessagePrioritizeProcessor -import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessagePrioritizationConstants -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) - - override suspend fun processNB(key: String, value: String) { - log.info("$$$$$ received in output processor key($key), value($value)") - val message = messagePrioritizationStateService.updateMessageState(value, MessageState.COMPLETED.name) - processorContext.forward(message.id, message, To.child(MessagePrioritizationConstants.SINK_OUTPUT)) - } -} 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 49230b6e4..186499d66 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 @@ -17,28 +17,27 @@ package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.utils import org.apache.kafka.streams.processor.ProcessorSupplier -import org.onap.ccsdk.cds.blueprintsprocessor.core.service.BluePrintClusterService +import org.onap.ccsdk.cds.blueprintsprocessor.atomix.optionalClusterService 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.kafka.AbstractMessagePrioritizeProcessor 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 object MessageProcessorUtils { - /** Utility to create the cluster lock for message [messagePrioritization] */ - suspend fun prioritizationGrouplock( - clusterService: BluePrintClusterService?, - messagePrioritization: MessagePrioritization - ): ClusterLock? { + /** Utility to create the cluster lock for message [messagePrioritization] prioritization procssing.*/ + suspend fun prioritizationGrouplock(messagePrioritization: MessagePrioritization): ClusterLock? { + val clusterService = BluePrintDependencyService.optionalClusterService() + 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 lockName = "prioritize::${messagePrioritization.group}::$correlationId" val clusterLock = clusterService.clusterLock(lockName) clusterLock.lock() if (!clusterLock.isLocked()) throw BluePrintProcessorException("failed to lock($lockName)") @@ -46,14 +45,15 @@ object MessageProcessorUtils { } else null } - /** Utility used to cluster unlock for message [messagePrioritization] */ - suspend fun prioritizationGroupUnLock(clusterService: BluePrintClusterService?, clusterLock: ClusterLock?) { - if (clusterService != null && clusterService.clusterJoined() && clusterLock != null) { + /** Utility used to cluster unlock for message [clusterLock] */ + suspend fun prioritizationGroupUnLock(clusterLock: ClusterLock?) { + if (clusterLock != null) { clusterLock.unLock() clusterLock.close() } } + /** Get the Kafka Supplier for processor lookup [name] and [prioritizationConfiguration] **/ fun <K, V> bluePrintProcessorSupplier(name: String, prioritizationConfiguration: PrioritizationConfiguration): ProcessorSupplier<K, V> { return ProcessorSupplier<K, V> { 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 f9e23e826..ec0515c42 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 @@ -27,12 +27,13 @@ import org.junit.runner.RunWith import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.PrioritizationMessageRepository -import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.service.MessagePrioritizationStateService +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.kafka.MessagePrioritizationConsumer import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.utils.MessagePrioritizationSample import org.onap.ccsdk.cds.blueprintsprocessor.message.BluePrintMessageLibConfiguration import org.onap.ccsdk.cds.blueprintsprocessor.message.service.BluePrintMessageLibPropertyService import org.onap.ccsdk.cds.blueprintsprocessor.message.service.KafkaBasicAuthMessageProducerService import org.onap.ccsdk.cds.controllerblueprints.core.asJsonString +import org.onap.ccsdk.cds.controllerblueprints.core.logger import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService import org.springframework.beans.factory.annotation.Autowired import org.springframework.boot.test.autoconfigure.orm.jpa.DataJpaTest @@ -43,6 +44,7 @@ import org.springframework.test.context.TestPropertySource import org.springframework.test.context.junit4.SpringRunner import kotlin.test.Test import kotlin.test.assertNotNull +import kotlin.test.assertTrue @RunWith(SpringRunner::class) @DataJpaTest @@ -72,6 +74,8 @@ import kotlin.test.assertNotNull ) open class MessagePrioritizationConsumerTest { + private val log = logger(MessagePrioritizationConsumerTest::class) + @Autowired lateinit var applicationContext: ApplicationContext @@ -82,6 +86,9 @@ open class MessagePrioritizationConsumerTest { lateinit var bluePrintMessageLibPropertyService: BluePrintMessageLibPropertyService @Autowired + lateinit var messagePrioritizationService: MessagePrioritizationService + + @Autowired lateinit var messagePrioritizationConsumer: MessagePrioritizationConsumer @Before @@ -107,6 +114,38 @@ open class MessagePrioritizationConsumerTest { } @Test + fun testMessagePrioritizationService() { + runBlocking { + assertTrue( + ::messagePrioritizationService.isInitialized, + "failed to initialize messagePrioritizationService" + ) + + log.info("**************** without Correlation **************") + /** Checking without correlation */ + MessagePrioritizationSample.sampleMessages(MessageState.NEW.name, 2).forEach { + messagePrioritizationService.prioritize(it) + } + log.info("**************** Same Group , with Correlation **************") + /** checking same group with correlation */ + MessagePrioritizationSample + .sampleMessageWithSameCorrelation("same-group", MessageState.NEW.name, 2) + .forEach { + delay(10) + messagePrioritizationService.prioritize(it) + } + log.info("**************** Different Type , with Correlation **************") + /** checking different type, with correlation */ + MessagePrioritizationSample + .sampleMessageWithDifferentTypeSameCorrelation("group-typed", MessageState.NEW.name, 3) + .forEach { + delay(10) + messagePrioritizationService.prioritize(it) + } + } + } + + @Test fun testStartConsuming() { runBlocking { val configuration = MessagePrioritizationSample.samplePrioritizationConfiguration() @@ -118,7 +157,9 @@ open class MessagePrioritizationConsumerTest { val spyStreamingConsumerService = spyk(streamingConsumerService) coEvery { spyStreamingConsumerService.consume(any(), any()) } returns Unit coEvery { spyStreamingConsumerService.shutDown() } returns Unit - val messagePrioritizationConsumer = MessagePrioritizationConsumer(bluePrintMessageLibPropertyService) + val messagePrioritizationConsumer = MessagePrioritizationConsumer( + bluePrintMessageLibPropertyService + ) val spyMessagePrioritizationConsumer = spyk(messagePrioritizationConsumer) // Test Topology 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 3d3d0c6f5..0285079ad 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 @@ -17,10 +17,9 @@ package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization import org.onap.ccsdk.cds.blueprintsprocessor.db.PrimaryDBLibGenericService -import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization -import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.topology.MessageAggregateProcessor -import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.topology.MessageOutputProcessor -import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.topology.MessagePrioritizeProcessor +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.kafka.DefaultMessagePrioritizeProcessor +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.kafka.MessagePrioritizationConsumer +import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.service.SampleMessagePrioritizationService import org.onap.ccsdk.cds.blueprintsprocessor.message.service.BluePrintMessageLibPropertyService import org.springframework.boot.autoconfigure.EnableAutoConfiguration import org.springframework.context.annotation.Bean @@ -65,22 +64,17 @@ open class SamplePrioritizationListeners(private val defaultMessagePrioritizatio */ @Service -open class SampleMessagePrioritizationConsumer( +open class TestMessagePrioritizationService(messagePrioritizationStateService: MessagePrioritizationStateService) : + SampleMessagePrioritizationService(messagePrioritizationStateService) + +/** For Kafka Consumer **/ +@Service +open class TestMessagePrioritizationConsumer( bluePrintMessageLibPropertyService: BluePrintMessageLibPropertyService ) : MessagePrioritizationConsumer(bluePrintMessageLibPropertyService) @Service(MessagePrioritizationConstants.PROCESSOR_PRIORITIZE) -open class SampleMessagePrioritizeProcessor : MessagePrioritizeProcessor() { - override fun getGroupCorrelationTypes(messagePrioritization: MessagePrioritization): List<String>? { - return when (messagePrioritization.group) { - "group-typed" -> arrayListOf("type-0", "type-1", "type-2") - else -> null - } - } -} - -@Service(MessagePrioritizationConstants.PROCESSOR_AGGREGATE) -open class SampleMessageAggregateProcessor() : MessageAggregateProcessor() - -@Service(MessagePrioritizationConstants.PROCESSOR_OUTPUT) -open class SampleMessageOutputProcessor : MessageOutputProcessor() +open class TestMessagePrioritizeProcessor( + messagePrioritizationStateService: MessagePrioritizationStateService, + messagePrioritizationService: MessagePrioritizationService +) : DefaultMessagePrioritizeProcessor(messagePrioritizationStateService, messagePrioritizationService) diff --git a/ms/blueprintsprocessor/modules/commons/atomix-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/atomix/BluePrintAtomixLibConfiguration.kt b/ms/blueprintsprocessor/modules/commons/atomix-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/atomix/BluePrintAtomixLibConfiguration.kt index 8ea15935f..8ef290303 100644 --- a/ms/blueprintsprocessor/modules/commons/atomix-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/atomix/BluePrintAtomixLibConfiguration.kt +++ b/ms/blueprintsprocessor/modules/commons/atomix-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/atomix/BluePrintAtomixLibConfiguration.kt @@ -18,6 +18,7 @@ package org.onap.ccsdk.cds.blueprintsprocessor.atomix import org.onap.ccsdk.cds.blueprintsprocessor.atomix.service.AtomixBluePrintClusterService import org.onap.ccsdk.cds.blueprintsprocessor.core.service.BluePrintClusterService +import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService import org.springframework.context.annotation.Configuration @@ -29,3 +30,10 @@ open class BluePrintAtomixLibConfiguration */ fun BluePrintDependencyService.clusterService(): BluePrintClusterService = instance(AtomixBluePrintClusterService::class) + +/** Optional Cluster Service, returns only if Cluster is enabled */ +fun BluePrintDependencyService.optionalClusterService(): BluePrintClusterService? { + return if (BluePrintConstants.CLUSTER_ENABLED) { + BluePrintDependencyService.clusterService() + } else null +} |