From 2dd658268202f92e0b5621093747f13670e5d042 Mon Sep 17 00:00:00 2001 From: mpriyank Date: Mon, 4 Nov 2024 15:59:58 +0000 Subject: Cm Avc Event to have same key - incoming Cm Avc Event from DMI Plugin is consumed and forwarded to target topic - the key from source topic to be used in the target topic while forwarding - with same key the ordering of the message will be preserved - NOTE: the RTD related changes will be a separate patchset Issue-ID: CPS-2436 Change-Id: Ie692663706b378022ec0d621d92ca5054bad8d1b Signed-off-by: mpriyank --- .../impl/cmnotificationsubscription/cmavc/CmAvcEventConsumer.java | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) (limited to 'cps-ncmp-service/src/main/java/org') diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/cmavc/CmAvcEventConsumer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/cmavc/CmAvcEventConsumer.java index 9e90eabbc4..2d1f64802b 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/cmavc/CmAvcEventConsumer.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/cmavc/CmAvcEventConsumer.java @@ -46,7 +46,8 @@ public class CmAvcEventConsumer { private final EventsPublisher eventsPublisher; /** - * Incoming AvcEvent in the form of Consumer Record. + * Incoming Cm AvcEvent in the form of Consumer Record, it will be forwarded as is to a target topic. + * The key from incoming record will be used as key for the target topic as well to preserve the message ordering. * * @param cmAvcEventAsConsumerRecord Incoming raw consumer record */ @@ -55,7 +56,8 @@ public class CmAvcEventConsumer { public void consumeAndForward( final ConsumerRecord cmAvcEventAsConsumerRecord) { final CloudEvent outgoingAvcEvent = cmAvcEventAsConsumerRecord.value(); - log.debug("Consuming AVC event {} ...", outgoingAvcEvent); - eventsPublisher.publishCloudEvent(cmEventsTopicName, outgoingAvcEvent.getId(), outgoingAvcEvent); + final String outgoingAvcEventKey = cmAvcEventAsConsumerRecord.key(); + log.debug("Consuming AVC event with key : {} and value : {}", outgoingAvcEventKey, outgoingAvcEvent); + eventsPublisher.publishCloudEvent(cmEventsTopicName, outgoingAvcEventKey, outgoingAvcEvent); } } -- cgit 1.2.3-korg