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 --- .../cmnotificationsubscription/cmavc/CmAvcEventConsumer.java | 8 +++++--- .../cmavc/CmAvcEventConsumerSpec.groovy | 11 +++++++---- .../cps/integration/functional/ncmp/CmHandleCreateSpec.groovy | 2 +- k6-tests/once-off-test/kafka/produce-avc-event.js | 8 +++++++- 4 files changed, 20 insertions(+), 9 deletions(-) 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); } } diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/cmnotificationsubscription/cmavc/CmAvcEventConsumerSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/cmnotificationsubscription/cmavc/CmAvcEventConsumerSpec.groovy index 06651be913..ad5f42ed94 100644 --- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/cmnotificationsubscription/cmavc/CmAvcEventConsumerSpec.groovy +++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/cmnotificationsubscription/cmavc/CmAvcEventConsumerSpec.groovy @@ -64,6 +64,7 @@ class CmAvcEventConsumerSpec extends MessagingBaseSpec { cloudEventKafkaConsumer.subscribe([cmEventsTopicName] as List) and: 'an event is sent' def jsonData = TestUtils.getResourceFileContent('sampleAvcInputEvent.json') + def testEventKey = 'sample-eventid-key' def testEventSent = jsonObjectMapper.convertJsonString(jsonData, AvcEvent.class) def testCloudEventSent = CloudEventBuilder.v1() .withData(jsonObjectMapper.asJsonBytes(testEventSent)) @@ -72,17 +73,19 @@ class CmAvcEventConsumerSpec extends MessagingBaseSpec { .withSource(URI.create('sample-test-source')) .withExtension('correlationid', 'test-cmhandle1').build() and: 'event has header information' - def consumerRecord = new ConsumerRecord(cmEventsTopicName, 0, 0, 'sample-eventid', testCloudEventSent) - when: 'the event is consumed' + def consumerRecord = new ConsumerRecord(cmEventsTopicName, 0, 0, testEventKey, testCloudEventSent) + when: 'the event is consumed and forwarded to target topic' acvEventConsumer.consumeAndForward(consumerRecord) - and: 'the topic is polled' + and: 'the target topic is polled' def records = cloudEventKafkaConsumer.poll(Duration.ofMillis(1500)) then: 'poll returns one record' assert records.size() == 1 - and: 'record can be converted to AVC event' + and: 'target record can be converted to AVC event' def record = records.iterator().next() def cloudEvent = record.value() as CloudEvent def convertedAvcEvent = toTargetEvent(cloudEvent, AvcEvent.class) + and: 'the target event has the same key as the source event to maintain the ordering in a partition' + assert record.key() == consumerRecord.key() and: 'we have correct headers forwarded where correlation id matches' assert KafkaHeaders.getParsedKafkaHeader(record.headers(), 'ce_correlationid') == 'test-cmhandle1' and: 'event id is same between consumed and forwarded' diff --git a/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/CmHandleCreateSpec.groovy b/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/CmHandleCreateSpec.groovy index 00ce38fa2d..ffcba025e8 100644 --- a/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/CmHandleCreateSpec.groovy +++ b/integration-test/src/test/groovy/org/onap/cps/integration/functional/ncmp/CmHandleCreateSpec.groovy @@ -242,7 +242,7 @@ class CmHandleCreateSpec extends CpsIntegrationSpecBase { if (retryAttempts == 0) break } - consumerRecords + return consumerRecords } } diff --git a/k6-tests/once-off-test/kafka/produce-avc-event.js b/k6-tests/once-off-test/kafka/produce-avc-event.js index 981a21af65..db222f6a4a 100644 --- a/k6-tests/once-off-test/kafka/produce-avc-event.js +++ b/k6-tests/once-off-test/kafka/produce-avc-event.js @@ -50,6 +50,11 @@ export const options = { } }; +const getRandomNetworkElement = () => { + const networkElementIds = Array.from({ length: 10 }, (_, i) => `neType-${i + 1}`); + return networkElementIds[Math.floor(Math.random() * networkElementIds.length)]; +}; + function getCloudEventHeaders() { return { ce_type: 'org.onap.cps.ncmp.events.avc1_0_0.AvcEvent', @@ -65,10 +70,11 @@ function getCloudEventHeaders() { export function sendKafkaMessages() { const cloudEventHeaders = getCloudEventHeaders(); + const networkElementId = getRandomNetworkElement(); const avcCloudEvent = { key: schemaRegistry.serialize({ - data: cloudEventHeaders.ce_correlationid, + data: networkElementId, schemaType: SCHEMA_TYPE_STRING, }), value: schemaRegistry.serialize({ -- cgit 1.2.3-korg