Replace event-id with correlation-id for data read operation cloud event

- Modified data operation event consumer to send correlation-id instead
  if event-id while publishing cloud event to client topic.

Issue-ID: CPS-1842
Signed-off-by: sourabh_sourabh <sourabh.sourabh@est.tech>
Change-Id: I30738b90d2ff0d14b5ee0bac680eff834aa968bf

2 files changed, 5 insertions, 4 deletions

diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/DataOperationEventConsumer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/DataOperationEventConsumer.java
index 9649b0109..0bc18076c 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/DataOperationEventConsumer.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/DataOperationEventConsumer.java
@@ -56,8 +56,8 @@ public class DataOperationEventConsumer {
         log.info("Consuming event payload {} ...", dataOperationEventConsumerRecord.value());
         final String eventTarget = KafkaHeaders.getParsedKafkaHeader(
                 dataOperationEventConsumerRecord.headers(), "ce_destination");
-        final String eventId = KafkaHeaders.getParsedKafkaHeader(
-                dataOperationEventConsumerRecord.headers(), "ce_id");
-        eventsPublisher.publishCloudEvent(eventTarget, eventId, dataOperationEventConsumerRecord.value());
+        final String correlationId = KafkaHeaders.getParsedKafkaHeader(
+                dataOperationEventConsumerRecord.headers(), "ce_correlationid");
+        eventsPublisher.publishCloudEvent(eventTarget, correlationId, dataOperationEventConsumerRecord.value());
     }
 }
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/SerializationIntegrationSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/SerializationIntegrationSpec.groovy
index 14ecd9282..f24ff6253 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/SerializationIntegrationSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/SerializationIntegrationSpec.groovy
@@ -68,7 +68,7 @@ class SerializationIntegrationSpec extends ConsumerBaseSpec {
         and: 'wait a little for async processing of message'
             TimeUnit.MILLISECONDS.sleep(300)
         then: 'the event has been forwarded'
-            1 * mockEventsPublisher.publishCloudEvent('some client topic', 'my-event-id', _) >> { args -> { capturedForwardedEvent = args[2] } }
+            1 * mockEventsPublisher.publishCloudEvent('some client topic', 'some-correlation-id', _) >> { args -> { capturedForwardedEvent = args[2] } }
         and: 'the forwarded event is identical to the event that was sent'
             assert capturedForwardedEvent == cloudEvent
     }
@@ -94,6 +94,7 @@ class SerializationIntegrationSpec extends ConsumerBaseSpec {
             .withType('DataOperationEvent')
             .withSource(URI.create('some-source'))
             .withExtension('destination','some client topic')
+            .withExtension('correlationid','some-correlation-id')
             .withData(objectMapper.writeValueAsBytes(dataOperationEvent))
             .build()
     }