NCMP : Handle non-existing and non-ready cm handles

- Modified data operation schema to contains cm handle as steing and
  response content field name as result.
- Added a new common cloud event builder for NCMP to create an event.
- Added data operation event creater that uses cloud event builder to
  create a cloud event.
- Introduced a new method onto json object mapper to convert json object
  to bytes.
- Modified EventDateTimeFormatter and added a new method to convert date timestamp into offsetdateTime.
- Added a new code into ResourceRequestUtil to identify non-ready cm
  handle and non-existing cm handle and later publish it as cloud event
  to tha client given topic.
- Introduced CpsApplicationContext to get spring mannaged bean into non
  spring managed java object.

Issue-ID: CPS-1583, CPS-1614

Signed-off-by: sourabh_sourabh <sourabh.sourabh@est.tech>
Change-Id: I24a39d2cb2c54dea25cd2f17e7748e21cd83a088
Signed-off-by: sourabh_sourabh <sourabh.sourabh@est.tech>

21 files changed, 664 insertions, 187 deletions

diff --git a/cps-ncmp-events/src/main/resources/schemas/async/data-operation-event-schema-1.0.0.json b/cps-ncmp-events/src/main/resources/schemas/async/data-operation-event-schema-1.0.0.json
index 308e3068d..f82e48141 100644
--- a/cps-ncmp-events/src/main/resources/schemas/async/data-operation-event-schema-1.0.0.json
+++ b/cps-ncmp-events/src/main/resources/schemas/async/data-operation-event-schema-1.0.0.json
@@ -19,12 +19,15 @@
                 "type": "object",
                 "properties": {
                   "operationId": {
-                    "description": "Used to distinguish multiple operations using same cmhandleId",
+                    "description": "Used to distinguish multiple operations using same handle ids",
                     "type": "string"
                   },
                   "ids": {
                     "description": "Id's of the cmhandles",
-                    "type": "array"
+                    "type": "array",
+                    "items": {
+                      "type": "string"
+                    }
                   },
                   "statusCode": {
                     "description": "which says success or failure (0-99) are for success and (100-199) are for failure",
@@ -34,7 +37,7 @@
                     "description": "Human readable message, Which says what the response has",
                     "type": "string"
                   },
-                  "responseContent": {
+                  "result": {
                     "description": "Contains the requested data response.",
                     "type": "object",
                     "existingJavaType": "java.lang.Object",
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/NcmpEventResponseCode.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/NcmpEventResponseCode.java
new file mode 100644
index 000000000..9f7ef1e88
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/NcmpEventResponseCode.java
@@ -0,0 +1,38 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2023 Nordix Foundation
+ *  ================================================================================
+ *  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.
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *  ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.api;
+
+import lombok.Getter;
+
+@Getter
+public enum NcmpEventResponseCode {
+
+    CODE_100("100", "cm handle id(s) not found"),
+    CODE_101("101", "cm handle(s) not ready");
+
+    private final String statusCode;
+    private final String statusMessage;
+
+    NcmpEventResponseCode(final String statusCode, final String statusMessage) {
+        this.statusCode = statusCode;
+        this.statusMessage = statusMessage;
+    }
+}
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/NcmpCloudEventBuilder.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/NcmpCloudEventBuilder.java
new file mode 100644
index 000000000..544db50a5
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/NcmpCloudEventBuilder.java
@@ -0,0 +1,64 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2023 Nordix Foundation
+ *  ================================================================================
+ *  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.
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *  ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.api.impl.events;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+import java.net.URI;
+import java.util.Map;
+import java.util.UUID;
+import lombok.Builder;
+import org.apache.commons.lang3.StringUtils;
+import org.onap.cps.ncmp.api.impl.utils.EventDateTimeFormatter;
+import org.onap.cps.ncmp.api.impl.utils.context.CpsApplicationContext;
+import org.onap.cps.utils.JsonObjectMapper;
+
+@Builder(buildMethodName = "setCloudEvent")
+public class NcmpCloudEventBuilder {
+
+    private Object event;
+    private Map<String, String> extensions;
+    private String type;
+    @Builder.Default
+    private static final String EVENT_SPEC_VERSION_V1 = "1.0.0";
+
+    /**
+     * Creates ncmp cloud event with provided attributes.
+     *
+     * @return Cloud Event
+     */
+    public CloudEvent build() {
+        final JsonObjectMapper jsonObjectMapper = CpsApplicationContext.getCpsBean(JsonObjectMapper.class);
+        final CloudEventBuilder cloudEventBuilder = CloudEventBuilder.v1()
+                .withId(UUID.randomUUID().toString())
+                .withSource(URI.create("NCMP"))
+                .withType(type)
+                .withDataSchema(URI.create("urn:cps:" + type + ":" + EVENT_SPEC_VERSION_V1))
+                .withTime(EventDateTimeFormatter.toIsoOffsetDateTime(
+                        EventDateTimeFormatter.getCurrentIsoFormattedDateTime()))
+                .withData(jsonObjectMapper.asJsonBytes(event));
+        extensions.entrySet().stream()
+                .filter(extensionEntry -> StringUtils.isNotBlank(extensionEntry.getValue()))
+                .forEach(extensionEntry ->
+                        cloudEventBuilder.withExtension(extensionEntry.getKey(), extensionEntry.getValue()));
+        return cloudEventBuilder.build();
+    }
+}
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/lcm/LcmEventsCreator.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/lcm/LcmEventsCreator.java
index 3c7c92b12..450bc8cce 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/lcm/LcmEventsCreator.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/lcm/LcmEventsCreator.java
@@ -108,7 +108,7 @@ public class LcmEventsCreator {
         final LcmEvent lcmEvent = new LcmEvent();
         lcmEvent.setEventId(UUID.randomUUID().toString());
         lcmEvent.setEventCorrelationId(eventCorrelationId);
-        lcmEvent.setEventTime(EventDateTimeFormatter.getCurrentDateTime());
+        lcmEvent.setEventTime(EventDateTimeFormatter.getCurrentIsoFormattedDateTime());
         lcmEvent.setEventSource("org.onap.ncmp");
         lcmEvent.setEventType(lcmEventType.getEventType());
         lcmEvent.setEventSchema("org.onap.ncmp:cmhandle-lcm-event");
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/operations/DmiDataOperations.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/operations/DmiDataOperations.java
index 8596c56dc..b4784f418 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/operations/DmiDataOperations.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/operations/DmiDataOperations.java
@@ -34,7 +34,7 @@ import org.onap.cps.ncmp.api.impl.client.DmiRestClient;
 import org.onap.cps.ncmp.api.impl.config.NcmpConfiguration;
 import org.onap.cps.ncmp.api.impl.executor.TaskExecutor;
 import org.onap.cps.ncmp.api.impl.utils.DmiServiceUrlBuilder;
-import org.onap.cps.ncmp.api.impl.utils.ResourceDataOperationRequestUtils;
+import org.onap.cps.ncmp.api.impl.utils.data.operation.ResourceDataOperationRequestUtils;
 import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle;
 import org.onap.cps.ncmp.api.inventory.CmHandleState;
 import org.onap.cps.ncmp.api.inventory.InventoryPersistence;
@@ -129,11 +129,11 @@ public class DmiDataOperations extends DmiOperations {
                 = getDistinctCmHandleIdsFromDataOperationRequest(dataOperationRequest);
 
         final Collection<YangModelCmHandle> yangModelCmHandles
-                = getYangModelCmHandlesInReadyState(cmHandlesIds);
+                = inventoryPersistence.getYangModelCmHandles(cmHandlesIds);
 
         final Map<String, List<DmiDataOperation>> operationsOutPerDmiServiceName
-                = ResourceDataOperationRequestUtils.processPerDefinitionInDataOperationsRequest(dataOperationRequest,
-                yangModelCmHandles);
+                = ResourceDataOperationRequestUtils.processPerDefinitionInDataOperationsRequest(topicParamInQuery,
+                requestId, dataOperationRequest, yangModelCmHandles);
 
         buildDataOperationRequestUrlAndSendToDmiService(topicParamInQuery, requestId, operationsOutPerDmiServiceName);
     }
@@ -221,15 +221,6 @@ public class DmiDataOperations extends DmiOperations {
                         dataOperationDefinition.getCmHandleIds().stream()).collect(Collectors.toSet());
     }
 
-    private Collection<YangModelCmHandle> getYangModelCmHandlesInReadyState(final Set<String> requestedCmHandleIds) {
-        // TODO Need to publish an error response to client given topic.
-        //  Code should be implemented into https://jira.onap.org/browse/CPS-1614 (
-        //  NCMP : Error handling for non-ready cm handle state)
-        return inventoryPersistence.getYangModelCmHandles(requestedCmHandleIds).stream()
-                .filter(yangModelCmHandle -> yangModelCmHandle.getCompositeState().getCmHandleState()
-                        == CmHandleState.READY).collect(Collectors.toList());
-    }
-
     private void buildDataOperationRequestUrlAndSendToDmiService(final String topicParamInQuery,
                                                                  final String requestId,
                                                                  final Map<String, List<DmiDataOperation>>
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/EventDateTimeFormatter.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/EventDateTimeFormatter.java
index acc4057d9..5dd682712 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/EventDateTimeFormatter.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/EventDateTimeFormatter.java
@@ -1,6 +1,6 @@
 /*
  * ============LICENSE_START=======================================================
- * Copyright (C) 2022 Nordix Foundation
+ * Copyright (C) 2022-2023 Nordix Foundation
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,23 +20,28 @@
 
 package org.onap.cps.ncmp.api.impl.utils;
 
+import java.time.OffsetDateTime;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
-import lombok.AccessLevel;
-import lombok.NoArgsConstructor;
+import org.apache.commons.lang3.StringUtils;
 
-@NoArgsConstructor(access = AccessLevel.PRIVATE)
-public class EventDateTimeFormatter {
+public interface EventDateTimeFormatter {
 
-    private static final String DATE_TIME_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSSZ";
+    String ISO_TIMESTAMP_PATTERN = "yyyy-MM-dd'T'HH:mm:ss.SSSZ";
+
+    DateTimeFormatter ISO_TIMESTAMP_FORMATTER = DateTimeFormatter.ofPattern(ISO_TIMESTAMP_PATTERN);
 
     /**
      * Gets current date time.
      *
      * @return the current date time
      */
-    public static String getCurrentDateTime() {
-        return ZonedDateTime.now()
-                .format(DateTimeFormatter.ofPattern(DATE_TIME_FORMAT));
+    static String getCurrentIsoFormattedDateTime() {
+        return ZonedDateTime.now().format(ISO_TIMESTAMP_FORMATTER);
+    }
+
+    static OffsetDateTime toIsoOffsetDateTime(final String dateTimestampAsString) {
+        return StringUtils.isNotBlank(dateTimestampAsString)
+                ? OffsetDateTime.parse(dateTimestampAsString, ISO_TIMESTAMP_FORMATTER) : null;
     }
 }
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/ResourceDataOperationRequestUtils.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/ResourceDataOperationRequestUtils.java
deleted file mode 100644
index 573f8b39a..000000000
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/ResourceDataOperationRequestUtils.java
+++ /dev/null
@@ -1,127 +0,0 @@
-/*
- *  ============LICENSE_START=======================================================
- *  Copyright (C) 2023 Nordix Foundation
- *  ================================================================================
- *  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.
- *
- *  SPDX-License-Identifier: Apache-2.0
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.cps.ncmp.api.impl.utils;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import lombok.AccessLevel;
-import lombok.NoArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import org.onap.cps.ncmp.api.impl.operations.CmHandle;
-import org.onap.cps.ncmp.api.impl.operations.DmiDataOperation;
-import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle;
-import org.onap.cps.ncmp.api.models.DataOperationDefinition;
-import org.onap.cps.ncmp.api.models.DataOperationRequest;
-
-@Slf4j
-@NoArgsConstructor(access = AccessLevel.PRIVATE)
-public class ResourceDataOperationRequestUtils {
-
-    private static final String UNKNOWN_SERVICE_NAME = null;
-
-    /**
-     * Create a list of DMI data operations per DMI service (name).
-     *
-     * @param dataOperationRequestIn incoming data operation request details
-     * @param yangModelCmHandles     involved cm handles represented as YangModelCmHandle (incl. metadata)
-     *
-     * @return {@code Map<String, List<DmiDataOperation>>} Create a list of DMI data operations operation
-     *                                                     per DMI service (name).
-     */
-    public static Map<String, List<DmiDataOperation>> processPerDefinitionInDataOperationsRequest(
-            final DataOperationRequest dataOperationRequestIn,
-            final Collection<YangModelCmHandle> yangModelCmHandles) {
-
-        final Map<String, Map<String, Map<String, String>>> dmiPropertiesPerCmHandleIdPerServiceName =
-                DmiServiceNameOrganizer.getDmiPropertiesPerCmHandleIdPerServiceName(yangModelCmHandles);
-
-        final Map<String, String> dmiServiceNamesPerCmHandleId =
-            getDmiServiceNamesPerCmHandleId(dmiPropertiesPerCmHandleIdPerServiceName);
-
-        final Map<String, List<DmiDataOperation>> dmiDataOperationsOutPerDmiServiceName = new HashMap<>();
-
-        for (final DataOperationDefinition dataOperationDefinitionIn :
-            dataOperationRequestIn.getDataOperationDefinitions()) {
-            for (final String cmHandleId : dataOperationDefinitionIn.getCmHandleIds()) {
-                final String dmiServiceName = dmiServiceNamesPerCmHandleId.get(cmHandleId);
-                final Map<String, String> cmHandleIdProperties
-                        = dmiPropertiesPerCmHandleIdPerServiceName.get(dmiServiceName).get(cmHandleId);
-                if (cmHandleIdProperties == null) {
-                    publishErrorMessageToClientTopic(cmHandleId);
-                } else {
-                    final DmiDataOperation dmiDataOperationOut = getOrAddDmiDataOperation(dmiServiceName,
-                            dataOperationDefinitionIn, dmiDataOperationsOutPerDmiServiceName);
-                    final CmHandle cmHandle = CmHandle.buildCmHandleWithProperties(cmHandleId, cmHandleIdProperties);
-                    dmiDataOperationOut.getCmHandles().add(cmHandle);
-                }
-            }
-        }
-        return dmiDataOperationsOutPerDmiServiceName;
-    }
-
-    private static void publishErrorMessageToClientTopic(final String requestedCmHandleId) {
-        log.warn("cm handle {} not found", requestedCmHandleId);
-        // TODO Need to publish an error response to client given topic.
-        //  Code should be implemented into https://jira.onap.org/browse/CPS-1583 (
-        //  NCMP : Handle non-existing cm handles)
-    }
-
-    private static Map<String, String> getDmiServiceNamesPerCmHandleId(
-            final Map<String, Map<String, Map<String, String>>> dmiDmiPropertiesPerCmHandleIdPerServiceName) {
-        final Map<String, String> dmiServiceNamesPerCmHandleId = new HashMap<>();
-        for (final Map.Entry<String, Map<String, Map<String, String>>> dmiDmiPropertiesEntry
-                : dmiDmiPropertiesPerCmHandleIdPerServiceName.entrySet()) {
-            final String dmiServiceName = dmiDmiPropertiesEntry.getKey();
-            final Set<String> cmHandleIds = dmiDmiPropertiesPerCmHandleIdPerServiceName.get(dmiServiceName).keySet();
-            for (final String cmHandleId : cmHandleIds) {
-                dmiServiceNamesPerCmHandleId.put(cmHandleId, dmiServiceName);
-            }
-        }
-        dmiDmiPropertiesPerCmHandleIdPerServiceName.put(UNKNOWN_SERVICE_NAME, Collections.emptyMap());
-        return dmiServiceNamesPerCmHandleId;
-    }
-
-    private static DmiDataOperation getOrAddDmiDataOperation(final String dmiServiceName,
-                                                             final DataOperationDefinition
-                                                                       dataOperationDefinitionIn,
-                                                             final Map<String, List<DmiDataOperation>>
-                                                                       dmiDataOperationsOutPerDmiServiceName) {
-        dmiDataOperationsOutPerDmiServiceName
-                .computeIfAbsent(dmiServiceName, dmiServiceNameAsKey -> new ArrayList<>());
-        final List<DmiDataOperation> dmiDataOperationsOut
-                = dmiDataOperationsOutPerDmiServiceName.get(dmiServiceName);
-        final boolean isNewOperation = dmiDataOperationsOut.isEmpty()
-                || !dmiDataOperationsOut.get(dmiDataOperationsOut.size() - 1).getOperationId()
-                .equals(dataOperationDefinitionIn.getOperationId());
-        if (isNewOperation) {
-            final DmiDataOperation newDmiDataOperationOut =
-                    DmiDataOperation.buildDmiDataOperationRequestBodyWithoutCmHandles(dataOperationDefinitionIn);
-            dmiDataOperationsOut.add(newDmiDataOperationOut);
-            return newDmiDataOperationOut;
-        }
-        return dmiDataOperationsOut.get(dmiDataOperationsOut.size() - 1);
-    }
-}
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/context/CpsApplicationContext.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/context/CpsApplicationContext.java
new file mode 100644
index 000000000..b14cf0d0d
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/context/CpsApplicationContext.java
@@ -0,0 +1,51 @@
+/*
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2023 Nordix Foundation
+ * ================================================================================
+ * 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.api.impl.utils.context;
+
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.stereotype.Component;
+
+@Component
+public class CpsApplicationContext implements ApplicationContextAware {
+
+    private static ApplicationContext applicationContext;
+
+    /**
+     * Returns the spring managed cps bean instance of the given class type if it exists.
+     * Returns null otherwise.
+     *
+     * @param cpsBeanClass cps class type
+     * @return requested bean instance
+     */
+    public static <T extends Object> T getCpsBean(final Class<T> cpsBeanClass) {
+        return applicationContext.getBean(cpsBeanClass);
+    }
+
+    @Override
+    public void setApplicationContext(final ApplicationContext cpsApplicationContext) {
+        setCpsApplicationContext(cpsApplicationContext);
+    }
+
+    private static synchronized void setCpsApplicationContext(final ApplicationContext cpsApplicationContext) {
+        CpsApplicationContext.applicationContext = cpsApplicationContext;
+    }
+}
\ No newline at end of file
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/data/operation/DataOperationEventCreator.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/data/operation/DataOperationEventCreator.java
new file mode 100644
index 000000000..2d9a51b84
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/data/operation/DataOperationEventCreator.java
@@ -0,0 +1,99 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2023 Nordix Foundation
+ *  ================================================================================
+ *  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.
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *  ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.api.impl.utils.data.operation;
+
+import io.cloudevents.CloudEvent;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.onap.cps.ncmp.api.NcmpEventResponseCode;
+import org.onap.cps.ncmp.api.impl.events.NcmpCloudEventBuilder;
+import org.onap.cps.ncmp.events.async1_0_0.Data;
+import org.onap.cps.ncmp.events.async1_0_0.DataOperationEvent;
+import org.onap.cps.ncmp.events.async1_0_0.Response;
+import org.springframework.util.MultiValueMap;
+
+@Slf4j
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class DataOperationEventCreator {
+
+    /**
+     * Creates data operation event.
+     *
+     * @param clientTopic                              topic the client wants to use for responses
+     * @param requestId                                unique identifier per request
+     * @param cmHandleIdsPerResponseCodesPerOperationId map of cm handles per operation response per response code
+     * @return Cloud Event
+     */
+    public static CloudEvent createDataOperationEvent(final String clientTopic,
+                                                      final String requestId,
+                                                      final MultiValueMap<String,
+                                                              Map<NcmpEventResponseCode, List<String>>>
+                                                              cmHandleIdsPerResponseCodesPerOperationId) {
+        final DataOperationEvent dataOperationEvent = new DataOperationEvent();
+        final Data data = createPayloadFromDataOperationResponses(cmHandleIdsPerResponseCodesPerOperationId);
+        dataOperationEvent.setData(data);
+        final Map<String, String> extensions = createDataOperationExtensions(requestId, clientTopic);
+        return NcmpCloudEventBuilder.builder().type(DataOperationEvent.class.getName())
+                .event(dataOperationEvent).extensions(extensions).setCloudEvent().build();
+    }
+
+    private static Data createPayloadFromDataOperationResponses(final MultiValueMap<String, Map<NcmpEventResponseCode,
+            List<String>>> cmHandleIdsPerOperationIdPerResponseCode) {
+        final Data data = new Data();
+        final List<org.onap.cps.ncmp.events.async1_0_0.Response> responses = new ArrayList<>();
+        cmHandleIdsPerOperationIdPerResponseCode.entrySet().forEach(cmHandleIdsPerOperationIdPerResponseCodeEntries ->
+                cmHandleIdsPerOperationIdPerResponseCodeEntries.getValue().forEach(cmHandleIdsPerResponseCodeEntries ->
+                        responses.addAll(createResponseFromDataOperationResponses(
+                                cmHandleIdsPerOperationIdPerResponseCodeEntries.getKey(),
+                                cmHandleIdsPerResponseCodeEntries)
+                        )));
+        data.setResponses(responses);
+        return data;
+    }
+
+    private static List<Response> createResponseFromDataOperationResponses(
+            final String operationId,
+            final Map<NcmpEventResponseCode, List<String>> cmHandleIdsPerResponseCodeEntries) {
+        final List<org.onap.cps.ncmp.events.async1_0_0.Response> responses = new ArrayList<>();
+        cmHandleIdsPerResponseCodeEntries.entrySet()
+                .forEach(cmHandleIdsPerResponseCodeEntry -> {
+                    final Response response = new Response();
+                    response.setOperationId(operationId);
+                    response.setStatusCode(cmHandleIdsPerResponseCodeEntry.getKey().getStatusCode());
+                    response.setStatusMessage(cmHandleIdsPerResponseCodeEntry.getKey().getStatusMessage());
+                    response.setIds(cmHandleIdsPerResponseCodeEntry.getValue());
+                    responses.add(response);
+                });
+        return responses;
+    }
+
+    private static Map<String, String> createDataOperationExtensions(final String requestId, final String clientTopic) {
+        final Map<String, String> extensions = new HashMap<>();
+        extensions.put("correlationid", requestId);
+        extensions.put("destination", clientTopic);
+        return extensions;
+    }
+}
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/data/operation/ResourceDataOperationRequestUtils.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/data/operation/ResourceDataOperationRequestUtils.java
new file mode 100644
index 000000000..957f48a86
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/data/operation/ResourceDataOperationRequestUtils.java
@@ -0,0 +1,178 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2023 Nordix Foundation
+ *  ================================================================================
+ *  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.
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.api.impl.utils.data.operation;
+
+import io.cloudevents.CloudEvent;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.onap.cps.ncmp.api.NcmpEventResponseCode;
+import org.onap.cps.ncmp.api.impl.events.EventsPublisher;
+import org.onap.cps.ncmp.api.impl.operations.CmHandle;
+import org.onap.cps.ncmp.api.impl.operations.DmiDataOperation;
+import org.onap.cps.ncmp.api.impl.utils.DmiServiceNameOrganizer;
+import org.onap.cps.ncmp.api.impl.utils.context.CpsApplicationContext;
+import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle;
+import org.onap.cps.ncmp.api.inventory.CmHandleState;
+import org.onap.cps.ncmp.api.models.DataOperationDefinition;
+import org.onap.cps.ncmp.api.models.DataOperationRequest;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+
+@Slf4j
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class ResourceDataOperationRequestUtils {
+
+    private static final String UNKNOWN_SERVICE_NAME = null;
+
+    /**
+     * Create a list of DMI data operation per DMI service (name).
+     *
+     * @param topicParamInQuery      client given topic
+     * @param requestId              unique identifier per request
+     * @param dataOperationRequestIn incoming data operation request details
+     * @param yangModelCmHandles     involved cm handles represented as YangModelCmHandle (incl. metadata)
+     * @return {@code Map<String, List<DmiBatchOperation>>} Create a list of DMI batch operation per DMI service (name).
+     */
+    public static Map<String, List<DmiDataOperation>> processPerDefinitionInDataOperationsRequest(
+            final String topicParamInQuery,
+            final String requestId,
+            final DataOperationRequest dataOperationRequestIn,
+            final Collection<YangModelCmHandle> yangModelCmHandles) {
+
+        final Map<String, List<DmiDataOperation>> dmiDataOperationsOutPerDmiServiceName = new HashMap<>();
+        final MultiValueMap<String, Map<NcmpEventResponseCode, List<String>>> cmHandleIdsPerOperationIdPerResponseCode
+                = new LinkedMultiValueMap<>();
+        final Set<String> nonReadyCmHandleIdsLookup = filterAndGetNonReadyCmHandleIds(yangModelCmHandles);
+
+        final Map<String, Map<String, Map<String, String>>> dmiPropertiesPerCmHandleIdPerServiceName =
+                DmiServiceNameOrganizer.getDmiPropertiesPerCmHandleIdPerServiceName(yangModelCmHandles);
+
+        final Map<String, String> dmiServiceNamesPerCmHandleId =
+                getDmiServiceNamesPerCmHandleId(dmiPropertiesPerCmHandleIdPerServiceName);
+
+        for (final DataOperationDefinition dataOperationDefinitionIn :
+                dataOperationRequestIn.getDataOperationDefinitions()) {
+            final List<String> nonExistingCmHandleIds = new ArrayList<>();
+            final List<String> nonReadyCmHandleIds = new ArrayList<>();
+            for (final String cmHandleId : dataOperationDefinitionIn.getCmHandleIds()) {
+                if (nonReadyCmHandleIdsLookup.contains(cmHandleId)) {
+                    nonReadyCmHandleIds.add(cmHandleId);
+                } else {
+                    final String dmiServiceName = dmiServiceNamesPerCmHandleId.get(cmHandleId);
+                    final Map<String, String> cmHandleIdProperties
+                            = dmiPropertiesPerCmHandleIdPerServiceName.get(dmiServiceName).get(cmHandleId);
+                    if (cmHandleIdProperties == null) {
+                        nonExistingCmHandleIds.add(cmHandleId);
+                    } else {
+                        final DmiDataOperation dmiBatchOperationOut = getOrAddDmiBatchOperation(dmiServiceName,
+                                dataOperationDefinitionIn, dmiDataOperationsOutPerDmiServiceName);
+                        final CmHandle cmHandle = CmHandle.buildCmHandleWithProperties(cmHandleId,
+                                cmHandleIdProperties);
+                        dmiBatchOperationOut.getCmHandles().add(cmHandle);
+                    }
+                }
+            }
+            populateCmHandleIdsPerOperationIdPerResponseCode(cmHandleIdsPerOperationIdPerResponseCode,
+                    dataOperationDefinitionIn.getOperationId(), NcmpEventResponseCode.CODE_100, nonExistingCmHandleIds);
+            populateCmHandleIdsPerOperationIdPerResponseCode(cmHandleIdsPerOperationIdPerResponseCode,
+                    dataOperationDefinitionIn.getOperationId(), NcmpEventResponseCode.CODE_101, nonReadyCmHandleIds);
+        }
+        if (!cmHandleIdsPerOperationIdPerResponseCode.isEmpty()) {
+            publishErrorMessageToClientTopic(topicParamInQuery, requestId, cmHandleIdsPerOperationIdPerResponseCode);
+        }
+        return dmiDataOperationsOutPerDmiServiceName;
+    }
+
+    @Async
+    private static void publishErrorMessageToClientTopic(final String clientTopic,
+                                                         final String requestId,
+                                                         final MultiValueMap<String,
+                                                                 Map<NcmpEventResponseCode, List<String>>>
+                                                                 cmHandleIdsPerOperationIdPerResponseCode) {
+        final CloudEvent dataOperationCloudEvent = DataOperationEventCreator.createDataOperationEvent(clientTopic,
+                requestId, cmHandleIdsPerOperationIdPerResponseCode);
+        final EventsPublisher<CloudEvent> eventsPublisher = CpsApplicationContext.getCpsBean(EventsPublisher.class);
+        eventsPublisher.publishCloudEvent(clientTopic, requestId, dataOperationCloudEvent);
+    }
+
+    private static Map<String, String> getDmiServiceNamesPerCmHandleId(
+            final Map<String, Map<String, Map<String, String>>> dmiDmiPropertiesPerCmHandleIdPerServiceName) {
+        final Map<String, String> dmiServiceNamesPerCmHandleId = new HashMap<>();
+        for (final Map.Entry<String, Map<String, Map<String, String>>> dmiDmiPropertiesEntry
+                : dmiDmiPropertiesPerCmHandleIdPerServiceName.entrySet()) {
+            final String dmiServiceName = dmiDmiPropertiesEntry.getKey();
+            final Set<String> cmHandleIds = dmiDmiPropertiesPerCmHandleIdPerServiceName.get(dmiServiceName).keySet();
+            for (final String cmHandleId : cmHandleIds) {
+                dmiServiceNamesPerCmHandleId.put(cmHandleId, dmiServiceName);
+            }
+        }
+        dmiDmiPropertiesPerCmHandleIdPerServiceName.put(UNKNOWN_SERVICE_NAME, Collections.emptyMap());
+        return dmiServiceNamesPerCmHandleId;
+    }
+
+    private static DmiDataOperation getOrAddDmiBatchOperation(final String dmiServiceName,
+                                                               final DataOperationDefinition
+                                                                       dataOperationDefinitionIn,
+                                                               final Map<String, List<DmiDataOperation>>
+                                                                       dmiBatchOperationsOutPerDmiServiceName) {
+        dmiBatchOperationsOutPerDmiServiceName
+                .computeIfAbsent(dmiServiceName, dmiServiceNameAsKey -> new ArrayList<>());
+        final List<DmiDataOperation> dmiBatchOperationsOut
+                = dmiBatchOperationsOutPerDmiServiceName.get(dmiServiceName);
+        final boolean isNewOperation = dmiBatchOperationsOut.isEmpty()
+                || !dmiBatchOperationsOut.get(dmiBatchOperationsOut.size() - 1).getOperationId()
+                .equals(dataOperationDefinitionIn.getOperationId());
+        if (isNewOperation) {
+            final DmiDataOperation newDmiBatchOperationOut =
+                    DmiDataOperation.buildDmiDataOperationRequestBodyWithoutCmHandles(dataOperationDefinitionIn);
+            dmiBatchOperationsOut.add(newDmiBatchOperationOut);
+            return newDmiBatchOperationOut;
+        }
+        return dmiBatchOperationsOut.get(dmiBatchOperationsOut.size() - 1);
+    }
+
+    private static Set<String> filterAndGetNonReadyCmHandleIds(final Collection<YangModelCmHandle> yangModelCmHandles) {
+        return yangModelCmHandles.stream()
+                .filter(yangModelCmHandle -> yangModelCmHandle.getCompositeState().getCmHandleState()
+                        != CmHandleState.READY).map(YangModelCmHandle::getId).collect(Collectors.toSet());
+    }
+
+    private static void populateCmHandleIdsPerOperationIdPerResponseCode(final MultiValueMap<String,
+            Map<NcmpEventResponseCode, List<String>>> cmHandleIdsPerOperationIdByResponseCode,
+                                                                        final String operationId,
+                                                                        final NcmpEventResponseCode
+                                                                                ncmpEventResponseCode,
+                                                                        final List<String> cmHandleIds) {
+        if (!cmHandleIds.isEmpty()) {
+            cmHandleIdsPerOperationIdByResponseCode.add(operationId, Map.of(ncmpEventResponseCode, cmHandleIds));
+        }
+    }
+}
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/NcmpAsyncDataOperationEventConsumerSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/NcmpAsyncDataOperationEventConsumerSpec.groovy
index 7f8469aaf..635328871 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/NcmpAsyncDataOperationEventConsumerSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/NcmpAsyncDataOperationEventConsumerSpec.groovy
@@ -92,7 +92,7 @@ class NcmpAsyncDataOperationEventConsumerSpec extends MessagingBaseSpec {
             response.operationId == 'some-operation-id'
             response.statusCode == 'any-success-status-code'
             response.statusMessage == 'Successfully applied changes'
-            response.responseContent as String == '[some-key:some-value]'
+            response.result as String == '[some-key:some-value]'
     }
 
     def 'Filter an event with type #eventType'() {
@@ -110,7 +110,7 @@ class NcmpAsyncDataOperationEventConsumerSpec extends MessagingBaseSpec {
 
     def createConsumerRecord(eventTypeAsString) {
         def jsonData = TestUtils.getResourceFileContent('dataOperationEvent.json')
-        def testEventSentAsBytes = objectMapper.writeValueAsBytes(jsonObjectMapper.convertJsonString(jsonData, DataOperationEvent.class))
+        def testEventSentAsBytes = jsonObjectMapper.asJsonBytes(jsonObjectMapper.convertJsonString(jsonData, DataOperationEvent.class))
 
         CloudEvent cloudEvent = getCloudEvent(eventTypeAsString, testEventSentAsBytes)
 
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/EventPublisherSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/EventPublisherSpec.groovy
new file mode 100644
index 000000000..59a43caf9
--- /dev/null
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/EventPublisherSpec.groovy
@@ -0,0 +1,86 @@
+/*
+ * ============LICENSE_START========================================================
+ * Copyright (c) 2023 Nordix Foundation.
+ *  ================================================================================
+ *  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.
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *  ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.api.impl.events
+
+import ch.qos.logback.classic.Level
+import ch.qos.logback.classic.Logger
+import ch.qos.logback.core.read.ListAppender
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.clients.producer.RecordMetadata
+import org.apache.kafka.common.TopicPartition
+import org.onap.cps.ncmp.init.SubscriptionModelLoader
+import org.slf4j.LoggerFactory
+import org.springframework.kafka.support.SendResult
+import spock.lang.Specification
+
+class EventPublisherSpec extends Specification {
+
+    def objectUnderTest = new EventsPublisher(null, null)
+    def logger = (Logger) LoggerFactory.getLogger(objectUnderTest.getClass())
+    def loggingListAppender
+
+    void setup() {
+        logger.setLevel(Level.DEBUG)
+        loggingListAppender = new ListAppender()
+        logger.addAppender(loggingListAppender)
+        loggingListAppender.start()
+    }
+
+    void cleanup() {
+        ((Logger) LoggerFactory.getLogger(SubscriptionModelLoader.class)).detachAndStopAllAppenders()
+    }
+
+    def 'Callback handling on success.'() {
+        given: 'a send result'
+            def producerRecord = new ProducerRecord('topic-1', 'my value')
+            def topicPartition = new TopicPartition('topic-2', 0)
+            def recordMetadata = new RecordMetadata(topicPartition, 0, 0, 0, 0, 0)
+            def sendResult = new SendResult(producerRecord, recordMetadata)
+        when: 'the callback handler processes success'
+            def callbackHandler = objectUnderTest.handleCallback('topic-3')
+            callbackHandler.onSuccess(sendResult)
+        then: 'an event is logged with level DEBUG'
+            def loggingEvent = getLoggingEvent()
+            loggingEvent.level == Level.DEBUG
+        and: 'it contains the topic (from the record metadata) and the "value" (from the producer record)'
+            loggingEvent.formattedMessage.contains('topic-2')
+            loggingEvent.formattedMessage.contains('my value')
+    }
+
+
+    def 'Callback handling on failure.'() {
+        when: 'the callback handler processes a failure'
+            def callbackHandler = objectUnderTest.handleCallback('my topic')
+            callbackHandler.onFailure(new Exception('my exception'))
+        then: 'an event is logged with level ERROR'
+            def loggingEvent = getLoggingEvent()
+            loggingEvent.level == Level.ERROR
+        and: 'it contains the topic and exception message'
+            loggingEvent.formattedMessage.contains('my topic')
+            loggingEvent.formattedMessage.contains('my exception')
+    }
+
+    def getLoggingEvent() {
+        return loggingListAppender.list[0]
+    }
+
+
+}
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avc/AvcEventConsumerSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avc/AvcEventConsumerSpec.groovy
index 5cc70e280..22852bea4 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avc/AvcEventConsumerSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avc/AvcEventConsumerSpec.groovy
@@ -55,9 +55,6 @@ class AvcEventConsumerSpec extends MessagingBaseSpec {
     @Autowired
     JsonObjectMapper jsonObjectMapper
 
-    @Autowired
-    ObjectMapper objectMapper
-
     def cloudEventKafkaConsumer = new KafkaConsumer<>(eventConsumerConfigProperties('ncmp-group', CloudEventDeserializer))
 
     def 'Consume and forward valid message'() {
@@ -69,7 +66,7 @@ class AvcEventConsumerSpec extends MessagingBaseSpec {
             def jsonData = TestUtils.getResourceFileContent('sampleAvcInputEvent.json')
             def testEventSent = jsonObjectMapper.convertJsonString(jsonData, AvcEvent.class)
             def testCloudEventSent = CloudEventBuilder.v1()
-                .withData(objectMapper.writeValueAsBytes(testEventSent))
+                .withData(jsonObjectMapper.asJsonBytes(testEventSent))
                 .withId('sample-eventid')
                 .withType('sample-test-type')
                 .withSource(URI.create('sample-test-source'))
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/operations/DmiDataOperationsSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/operations/DmiDataOperationsSpec.groovy
index c7ee4e074..59e62e34d 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/operations/DmiDataOperationsSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/operations/DmiDataOperationsSpec.groovy
@@ -23,8 +23,11 @@ package org.onap.cps.ncmp.api.impl.operations
 
 import com.fasterxml.jackson.databind.ObjectMapper
 import org.onap.cps.ncmp.api.impl.config.NcmpConfiguration
+import org.onap.cps.ncmp.api.impl.events.EventsPublisher
 import org.onap.cps.ncmp.api.impl.utils.DmiServiceUrlBuilder
+import org.onap.cps.ncmp.api.impl.utils.context.CpsApplicationContext
 import org.onap.cps.ncmp.api.models.DataOperationRequest
+import org.onap.cps.ncmp.event.model.NcmpAsyncRequestResponseEvent
 import org.onap.cps.ncmp.utils.TestUtils
 import org.onap.cps.utils.JsonObjectMapper
 import org.spockframework.spring.SpringBean
@@ -42,7 +45,7 @@ import static org.onap.cps.ncmp.api.impl.operations.OperationType.READ
 import static org.onap.cps.ncmp.api.impl.operations.OperationType.UPDATE
 
 @SpringBootTest
-@ContextConfiguration(classes = [NcmpConfiguration.DmiProperties, DmiDataOperations])
+@ContextConfiguration(classes = [EventsPublisher, CpsApplicationContext, NcmpConfiguration.DmiProperties, DmiDataOperations])
 class DmiDataOperationsSpec extends DmiOperationsBaseSpec {
 
     @SpringBean
@@ -59,6 +62,9 @@ class DmiDataOperationsSpec extends DmiOperationsBaseSpec {
     @Autowired
     DmiDataOperations objectUnderTest
 
+    @SpringBean
+    EventsPublisher eventsPublisher = Stub()
+
     def 'call get resource data for #expectedDatastoreInUrl from DMI without topic #scenario.'() {
         given: 'a cm handle for #cmHandleId'
             mockYangModelCmHandleRetrieval(dmiProperties)
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/utils/context/CpsApplicationContextSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/utils/context/CpsApplicationContextSpec.groovy
new file mode 100644
index 000000000..b7fa44925
--- /dev/null
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/utils/context/CpsApplicationContextSpec.groovy
@@ -0,0 +1,19 @@
+package org.onap.cps.ncmp.api.impl.utils.context
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import org.onap.cps.utils.JsonObjectMapper
+import org.springframework.boot.test.context.SpringBootTest
+import org.springframework.test.context.ContextConfiguration
+import spock.lang.Specification;
+
+@SpringBootTest(classes = [ObjectMapper, JsonObjectMapper])
+@ContextConfiguration(classes = [CpsApplicationContext.class])
+class CpsApplicationContextSpec extends Specification {
+
+    def 'Verify if cps application context contains a requested bean.'() {
+        when: 'cps bean is requested from application context'
+            def jsonObjectMapper = CpsApplicationContext.getCpsBean(JsonObjectMapper.class)
+        then: 'requested bean of JsonObjectMapper is not null'
+            assert jsonObjectMapper != null
+    }
+}
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/utils/DataOperationRequestUtilsSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/utils/data/operation/ResourceDataOperationRequestUtilsSpec.groovy
index 334b455ef..401254f54 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/utils/DataOperationRequestUtilsSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/utils/data/operation/ResourceDataOperationRequestUtilsSpec.groovy
@@ -18,23 +18,46 @@
  *  ============LICENSE_END=========================================================
  */
 
-package org.onap.cps.ncmp.api.impl.utils
+package org.onap.cps.ncmp.api.impl.utils.data.operation
 
 import com.fasterxml.jackson.databind.ObjectMapper
+import io.cloudevents.CloudEvent
+import io.cloudevents.core.CloudEventUtils
+import io.cloudevents.jackson.PojoCloudEventDataMapper
+import io.cloudevents.kafka.CloudEventDeserializer
+import io.cloudevents.kafka.impl.KafkaHeaders
+import org.apache.kafka.clients.consumer.KafkaConsumer
+import org.onap.cps.ncmp.api.impl.events.EventsPublisher
+import org.onap.cps.ncmp.api.impl.utils.context.CpsApplicationContext
 import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle
 import org.onap.cps.ncmp.api.inventory.CmHandleState
 import org.onap.cps.ncmp.api.inventory.CompositeStateBuilder
+import org.onap.cps.ncmp.api.kafka.MessagingBaseSpec
 import org.onap.cps.ncmp.api.models.DataOperationRequest
+import org.onap.cps.ncmp.events.async1_0_0.DataOperationEvent
 import org.onap.cps.ncmp.utils.TestUtils
 import org.onap.cps.utils.JsonObjectMapper
 import org.spockframework.spring.SpringBean
-import spock.lang.Specification
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.test.context.ContextConfiguration
+import java.time.Duration
 
-class DataOperationRequestUtilsSpec extends Specification {
+@ContextConfiguration(classes = [EventsPublisher, CpsApplicationContext, ObjectMapper])
+class ResourceDataOperationRequestUtilsSpec extends MessagingBaseSpec {
+
+    def cloudEventKafkaConsumer = new KafkaConsumer<>(eventConsumerConfigProperties('test', CloudEventDeserializer))
+    def static clientTopic = 'my-topic-name'
+    def static dataOperationType = 'org.onap.cps.ncmp.events.async1_0_0.DataOperationEvent'
 
     @SpringBean
     JsonObjectMapper jsonObjectMapper = new JsonObjectMapper(new ObjectMapper())
 
+    @SpringBean
+    EventsPublisher eventPublisher = new EventsPublisher<CloudEvent>(legacyEventKafkaTemplate, cloudEventKafkaTemplate)
+
+    @Autowired
+    ObjectMapper objectMapper
+
     def 'Process per data operation request with #serviceName.'() {
         given: 'data operation request with 3 operations'
             def dataOperationRequestJsonData = TestUtils.getResourceFileContent('dataOperationRequest.json')
@@ -42,7 +65,7 @@ class DataOperationRequestUtilsSpec extends Specification {
         and: '4 known cm handles: ch1-dmi1, ch2-dmi1, ch3-dmi2, ch4-dmi2'
             def yangModelCmHandles = getYangModelCmHandles()
         when: 'data operation request is processed'
-            def operationsOutPerDmiServiceName = ResourceDataOperationRequestUtils.processPerDefinitionInDataOperationsRequest(dataOperationRequest, yangModelCmHandles)
+            def operationsOutPerDmiServiceName = ResourceDataOperationRequestUtils.processPerDefinitionInDataOperationsRequest(clientTopic,'request-id', dataOperationRequest, yangModelCmHandles)
         and: 'converted to a json node'
             def dmiDataOperationRequestBody = jsonObjectMapper.asJsonString(operationsOutPerDmiServiceName.get(serviceName))
             def dmiDataOperationRequestBodyAsJsonNode = jsonObjectMapper.convertToJsonNode(dmiDataOperationRequestBody).get(operationIndex)
@@ -65,9 +88,37 @@ class DataOperationRequestUtilsSpec extends Specification {
             'dmi2'      | 2              || 'operational-15'    | 'ncmp-datastore:passthrough-operational' | ['ch4-dmi2']
     }
 
+    def 'Process per data operation request with non-ready, non-existing cm handle and publish event to client specified topic'() {
+        given: 'consumer subscribing to client topic'
+            cloudEventKafkaConsumer.subscribe([clientTopic])
+        and: 'data operation request having non-ready and non-existing cm handle ids'
+            def dataOperationRequestJsonData = TestUtils.getResourceFileContent('dataOperationRequest.json')
+            def dataOperationRequest = jsonObjectMapper.convertJsonString(dataOperationRequestJsonData, DataOperationRequest.class)
+        when: 'data operation request is processed'
+            ResourceDataOperationRequestUtils.processPerDefinitionInDataOperationsRequest(clientTopic, 'request-id', dataOperationRequest, yangModelCmHandles)
+        and: 'subscribed client specified topic is polled and first record is selected'
+            def consumerRecordOut = cloudEventKafkaConsumer.poll(Duration.ofMillis(1500))[0]
+        then: 'verify cloud compliant headers'
+            def consumerRecordOutHeaders = consumerRecordOut.headers()
+            assert KafkaHeaders.getParsedKafkaHeader(consumerRecordOutHeaders, 'ce_id') != null
+            assert KafkaHeaders.getParsedKafkaHeader(consumerRecordOutHeaders, 'ce_type') == dataOperationType
+        and: 'verify that extension is included into header'
+            assert KafkaHeaders.getParsedKafkaHeader(consumerRecordOutHeaders, 'ce_correlationid') == 'request-id'
+            assert KafkaHeaders.getParsedKafkaHeader(consumerRecordOutHeaders, 'ce_destination') == clientTopic
+        and: 'map consumer record to expected event type'
+            def dataOperationResponseEvent = CloudEventUtils.mapData(consumerRecordOut.value(),
+                    PojoCloudEventDataMapper.from(objectMapper, DataOperationEvent.class)).getValue()
+        and: 'data operation response event response size is 3'
+            dataOperationResponseEvent.data.responses.size() == 3
+        and: 'verify published response data as json string'
+            jsonObjectMapper.asJsonString(dataOperationResponseEvent.data.responses)
+                    == '[{"operationId":"operational-14","ids":["unknown-cm-handle"],"statusCode":"100","statusMessage":"cm handle id(s) not found"},{"operationId":"operational-14","ids":["non-ready-cm handle"],"statusCode":"101","statusMessage":"cm handle(s) not ready"},{"operationId":"running-12","ids":["non-ready-cm handle"],"statusCode":"101","statusMessage":"cm handle(s) not ready"}]'
+    }
+
     static def getYangModelCmHandles() {
         def dmiProperties = [new YangModelCmHandle.Property('prop', 'some DMI property')]
         def readyState = new CompositeStateBuilder().withCmHandleState(CmHandleState.READY).withLastUpdatedTimeNow().build()
+        def advisedState = new CompositeStateBuilder().withCmHandleState(CmHandleState.ADVISED).withLastUpdatedTimeNow().build()
         return [new YangModelCmHandle(id: 'ch1-dmi1', dmiServiceName: 'dmi1', dmiProperties: dmiProperties, compositeState: readyState),
                 new YangModelCmHandle(id: 'ch2-dmi1', dmiServiceName: 'dmi1', dmiProperties: dmiProperties, compositeState: readyState),
                 new YangModelCmHandle(id: 'ch6-dmi1', dmiServiceName: 'dmi1', dmiProperties: dmiProperties, compositeState: readyState),
@@ -75,6 +126,7 @@ class DataOperationRequestUtilsSpec extends Specification {
                 new YangModelCmHandle(id: 'ch3-dmi2', dmiServiceName: 'dmi2', dmiProperties: dmiProperties, compositeState: readyState),
                 new YangModelCmHandle(id: 'ch4-dmi2', dmiServiceName: 'dmi2', dmiProperties: dmiProperties, compositeState: readyState),
                 new YangModelCmHandle(id: 'ch7-dmi2', dmiServiceName: 'dmi2', dmiProperties: dmiProperties, compositeState: readyState),
+                new YangModelCmHandle(id: 'non-ready-cm handle', dmiServiceName: 'dmi2', dmiProperties: dmiProperties, compositeState: advisedState)
         ]
     }
 }
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/init/SubscriptionModelLoaderSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/init/SubscriptionModelLoaderSpec.groovy
index a14a0f286..b4e7813db 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/init/SubscriptionModelLoaderSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/init/SubscriptionModelLoaderSpec.groovy
@@ -23,8 +23,6 @@ package org.onap.cps.ncmp.init
 import ch.qos.logback.classic.Level
 import ch.qos.logback.classic.Logger
 import ch.qos.logback.core.read.ListAppender
-import org.junit.jupiter.api.AfterEach
-import org.junit.jupiter.api.BeforeEach
 import org.onap.cps.api.CpsAdminService
 import org.onap.cps.api.CpsDataService
 import org.onap.cps.api.CpsModuleService
@@ -53,22 +51,19 @@ class SubscriptionModelLoaderSpec extends Specification {
     def applicationReadyEvent = new ApplicationReadyEvent(new SpringApplication(), null, applicationContext, null)
 
     def yangResourceToContentMap
-    def logger
-    def appender
+    def logger = (Logger) LoggerFactory.getLogger(objectUnderTest.getClass())
+    def loggingListAppender
 
-    @BeforeEach
     void setup() {
         yangResourceToContentMap = objectUnderTest.createYangResourceToContentMap()
-        logger = (Logger) LoggerFactory.getLogger(objectUnderTest.getClass())
-        appender = new ListAppender()
         logger.setLevel(Level.DEBUG)
-        appender.start()
-        logger.addAppender(appender)
+        loggingListAppender = new ListAppender()
+        logger.addAppender(loggingListAppender)
+        loggingListAppender.start()
         applicationContext.refresh()
     }
 
-    @AfterEach
-    void teardown() {
+    void cleanup() {
         ((Logger) LoggerFactory.getLogger(SubscriptionModelLoader.class)).detachAndStopAllAppenders()
         applicationContext.close()
     }
@@ -123,7 +118,7 @@ class SubscriptionModelLoaderSpec extends Specification {
         and: 'the data service to create a top level datanode was not called'
             0 * mockCpsDataService.saveData(*_)
         and: 'the log message contains the correct exception message'
-            def logs = appender.list.toString()
+            def logs = loggingListAppender.list.toString()
             assert logs.contains("Retrieval of NCMP dataspace fails")
     }
 
@@ -168,7 +163,7 @@ class SubscriptionModelLoaderSpec extends Specification {
         when: 'the method to onboard model is called'
             objectUnderTest.onboardSubscriptionModel(yangResourceToContentMap)
         then: 'the log message contains the correct exception message'
-            def debugMessage = appender.list[0].toString()
+            def debugMessage = loggingListAppender.list[0].toString()
             assert debugMessage.contains("Creating schema set failed")
         and: 'exception is thrown'
             thrown(NcmpStartUpException)
@@ -183,7 +178,7 @@ class SubscriptionModelLoaderSpec extends Specification {
         then: 'no exception thrown'
             noExceptionThrown()
         and: 'the log message contains the correct exception message'
-            def infoMessage = appender.list[0].toString()
+            def infoMessage = loggingListAppender.list[0].toString()
             assert infoMessage.contains("already exists")
     }
 
@@ -194,7 +189,7 @@ class SubscriptionModelLoaderSpec extends Specification {
         when: 'the method to onboard model is called'
             objectUnderTest.onboardSubscriptionModel(yangResourceToContentMap)
         then: 'the log message contains the correct exception message'
-            def debugMessage = appender.list[0].toString()
+            def debugMessage = loggingListAppender.list[0].toString()
             assert debugMessage.contains("Schema Set not found")
         and: 'exception is thrown'
             thrown(NcmpStartUpException)
@@ -209,7 +204,7 @@ class SubscriptionModelLoaderSpec extends Specification {
         then: 'no exception thrown'
             noExceptionThrown()
         and: 'the log message contains the correct exception message'
-            def infoMessage = appender.list[0].toString()
+            def infoMessage = loggingListAppender.list[0].toString()
             assert infoMessage.contains("already exists")
     }
 
@@ -220,7 +215,7 @@ class SubscriptionModelLoaderSpec extends Specification {
         when: 'the method to onboard model is called'
             objectUnderTest.onboardSubscriptionModel(yangResourceToContentMap)
         then: 'the log message contains the correct exception message'
-            def debugMessage = appender.list[0].toString()
+            def debugMessage = loggingListAppender.list[0].toString()
             assert debugMessage.contains("Creating data node for subscription model failed: Invalid JSON")
         and: 'exception is thrown'
             thrown(NcmpStartUpException)
diff --git a/cps-ncmp-service/src/test/resources/dataOperationEvent.json b/cps-ncmp-service/src/test/resources/dataOperationEvent.json
index 0a32f38c0..08a58b39b 100644
--- a/cps-ncmp-service/src/test/resources/dataOperationEvent.json
+++ b/cps-ncmp-service/src/test/resources/dataOperationEvent.json
@@ -8,7 +8,7 @@
         ],
         "statusCode": "any-success-status-code",
         "statusMessage": "Successfully applied changes",
-        "responseContent": {
+        "result": {
           "some-key": "some-value"
         }
       }
diff --git a/cps-ncmp-service/src/test/resources/dataOperationRequest.json b/cps-ncmp-service/src/test/resources/dataOperationRequest.json
index 98ed39b9a..d2e0d6489 100644
--- a/cps-ncmp-service/src/test/resources/dataOperationRequest.json
+++ b/cps-ncmp-service/src/test/resources/dataOperationRequest.json
@@ -9,7 +9,8 @@
       "targetIds": [
         "ch3-dmi2",
         "unknown-cm-handle",
-        "ch6-dmi1"
+        "ch6-dmi1",
+        "non-ready-cm handle"
       ]
     },
     {
@@ -19,7 +20,8 @@
       "targetIds": [
         "ch1-dmi1",
         "ch7-dmi2",
-        "ch2-dmi1"
+        "ch2-dmi1",
+        "non-ready-cm handle"
       ]
     },
     {
diff --git a/cps-service/src/main/java/org/onap/cps/utils/JsonObjectMapper.java b/cps-service/src/main/java/org/onap/cps/utils/JsonObjectMapper.java
index 338a841a7..60a6e16fe 100644
--- a/cps-service/src/main/java/org/onap/cps/utils/JsonObjectMapper.java
+++ b/cps-service/src/main/java/org/onap/cps/utils/JsonObjectMapper.java
@@ -90,6 +90,22 @@ public class JsonObjectMapper {
     }
 
     /**
+     * Serializing generic json object to bytes using Jackson.
+     *
+     * @param jsonObject any json object value
+     * @return the generated JSON as a byte array.
+     */
+    public byte[] asJsonBytes(final Object jsonObject) {
+        try {
+            return objectMapper.writeValueAsBytes(jsonObject);
+        } catch (final JsonProcessingException jsonProcessingException) {
+            log.error("Parsing error occurred while converting JSON object to bytes.");
+            throw new DataValidationException("Parsing error occurred while converting given JSON object to bytes.",
+                    jsonProcessingException.getMessage());
+        }
+    }
+
+    /**
      * Deserialize JSON content from given JSON content String to JsonNode.
      *
      * @param jsonContent   JSON content
diff --git a/cps-service/src/test/groovy/org/onap/cps/utils/JsonObjectMapperSpec.groovy b/cps-service/src/test/groovy/org/onap/cps/utils/JsonObjectMapperSpec.groovy
index b70c43795..2332282e2 100644
--- a/cps-service/src/test/groovy/org/onap/cps/utils/JsonObjectMapperSpec.groovy
+++ b/cps-service/src/test/groovy/org/onap/cps/utils/JsonObjectMapperSpec.groovy
@@ -33,15 +33,17 @@ class JsonObjectMapperSpec extends Specification {
     def spiedObjectMapper = Spy(ObjectMapper)
     def jsonObjectMapper = new JsonObjectMapper(spiedObjectMapper)
 
-    def 'Map a structured object to json String.'() {
+    def 'Map a structured object to json #type.'() {
         given: 'an object model'
             def object = spiedObjectMapper.readValue(TestUtils.getResourceFileContent('bookstore.json'), Object)
         when: 'the object is mapped to string'
-            def content = jsonObjectMapper.asJsonString(object);
+            def content = type == 'String' ? jsonObjectMapper.asJsonString(object) : jsonObjectMapper.asJsonBytes(object)
         then: 'the result is a valid json string (can be parsed)'
-            def contentMap = new JsonSlurper().parseText(content)
+            def contentMap = new JsonSlurper().parseText(new String(content))
         and: 'the parsed content is as expected'
             assert contentMap.'test:bookstore'.'bookstore-name' == 'Chapters/Easons'
+        where: 'the following data stores are used'
+            type << ['String', 'bytes']
     }
 
     def 'Map a structured object to json String error.'() {