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>

9 files changed, 449 insertions, 150 deletions

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 0000000000..9f7ef1e882
--- /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 0000000000..544db50a55
--- /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 3c7c92b129..450bc8cce3 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 8596c56dca..b4784f418f 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 acc4057d9d..5dd6827126 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 573f8b39a9..0000000000
--- 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 0000000000..b14cf0d0db
--- /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 0000000000..2d9a51b844
--- /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 0000000000..957f48a862
--- /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));
+        }
+    }
+}