Spliting a data-job into sub-jobs for DMI Plugin

    - algorithm for create sub-job requests
    - added new method to DmiServiceUrlBuilder to get the write job url
    - created WriteOperationExaminer, DmiSubJobClient & testware

Issue-ID: CPS-2142
Change-Id: I258d334ef346cd388341a1deb4078d24d8bdb7cc
Signed-off-by: leventecsanyi <levente.csanyi@est.tech>

13 files changed, 501 insertions, 26 deletions

diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/datajobs/DataJobService.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/datajobs/DataJobService.java
index f22124593e..6ff79a9344 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/datajobs/DataJobService.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/datajobs/DataJobService.java
@@ -20,9 +20,11 @@
 
 package org.onap.cps.ncmp.api.datajobs;
 
+import java.util.List;
 import org.onap.cps.ncmp.api.datajobs.models.DataJobMetadata;
 import org.onap.cps.ncmp.api.datajobs.models.DataJobReadRequest;
 import org.onap.cps.ncmp.api.datajobs.models.DataJobWriteRequest;
+import org.onap.cps.ncmp.api.datajobs.models.SubJobWriteResponse;
 
 public interface DataJobService {
 
@@ -41,6 +43,8 @@ public interface DataJobService {
      * @param dataJobId           Unique identifier of the job within the request
      * @param dataJobMetadata     data job request headers
      * @param dataJobWriteRequest write data job request
+     * @return a list of sub-job write responses
      */
-    void writeDataJob(String dataJobId, DataJobMetadata dataJobMetadata, DataJobWriteRequest dataJobWriteRequest);
+    List<SubJobWriteResponse> writeDataJob(String dataJobId, DataJobMetadata dataJobMetadata,
+                                           DataJobWriteRequest dataJobWriteRequest);
 }
\ No newline at end of file
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/datajobs/models/DmiWriteOperation.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/datajobs/models/DmiWriteOperation.java
new file mode 100644
index 0000000000..7e9ca7988b
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/datajobs/models/DmiWriteOperation.java
@@ -0,0 +1,43 @@
+/*
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2024 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.datajobs.models;
+
+import java.util.Map;
+
+/**
+ * Describes the write data job operation to be forwarded to dmi.
+ *
+ * @param path               Identifier of a managed object (MO) on a network element. Defines the resource on which
+ *                           operation is executed. Typically, is Fully Distinguished Name (FDN).
+ * @param op                 Describes the operation to execute.  The value can be as below:
+ *                           e.g. "add", "replace", "remove", "action" etc.
+ * @param moduleSetTag       The module set tag of the CM Handle.
+ * @param value              The value to be written depends on the type of operation.
+ * @param operationId        Unique identifier of the operation within the request.
+ * @param privateProperties  Contains the private properties of a Cm Handle.
+ */
+public record DmiWriteOperation(
+        String path,
+        String op,
+        String moduleSetTag,
+        Object value,
+        String operationId,
+        Map<String, String> privateProperties) {}
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/datajobs/models/ProducerKey.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/datajobs/models/ProducerKey.java
new file mode 100644
index 0000000000..ac6b7f8622
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/datajobs/models/ProducerKey.java
@@ -0,0 +1,36 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2024 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.datajobs.models;
+
+/**
+ * Composite key created from the DMI Service name and a data producer identifier.
+ * Helps to group of the sub job request for a given DMI Plugin.
+ *
+ * @param dmiServiceName          Describes the name of the relevant DMI service.
+ * @param dataProducerIdentifier  The name of a data producer identifier from a Cm Handle.
+ */
+public record ProducerKey(String dmiServiceName, String dataProducerIdentifier) {
+
+    @Override
+    public String toString() {
+        return dmiServiceName + "#"  + dataProducerIdentifier;
+    }
+}
\ No newline at end of file
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/datajobs/models/SubJobWriteRequest.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/datajobs/models/SubJobWriteRequest.java
new file mode 100644
index 0000000000..432b21b8c0
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/datajobs/models/SubJobWriteRequest.java
@@ -0,0 +1,41 @@
+/*
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2024 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.datajobs.models;
+
+import java.util.Collection;
+
+/**
+ * Response data for a write operation by the DMI Plugin.
+ *
+ * @param dataAcceptType  Define the data response accept type.
+ *                        e.g. "application/vnd.3gpp.object-tree-hierarchical+json",
+ *                        "application/vnd.3gpp.object-tree-flat+json" etc.
+ * @param dataContentType Define the data request content type.
+ *                        e.g. "application/3gpp-json-patch+json" etc.
+ * @param dataProducerId  Identifier of the data producer.
+ *
+ * @param data            A collection of outgoing write operations.
+ */
+public record SubJobWriteRequest (
+        String dataAcceptType,
+        String dataContentType,
+        String dataProducerId,
+        Collection<DmiWriteOperation> data) {}
\ No newline at end of file
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/datajobs/models/SubJobWriteResponse.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/datajobs/models/SubJobWriteResponse.java
new file mode 100644
index 0000000000..9cdd8e0590
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/datajobs/models/SubJobWriteResponse.java
@@ -0,0 +1,30 @@
+/*
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2024 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.datajobs.models;
+
+/**
+ * Request data for a write operation towards DMI Plugin.
+ *
+ * @param subJobId        Identifier of the sub-job from DMI.
+ * @param dmiServiceName  The provided name of the DMI service from the request.
+ * @param dataProducerId  Identifier of the data producer.
+ */
+public record SubJobWriteResponse(String subJobId, String dmiServiceName, String dataProducerId) {}
\ No newline at end of file
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/datajobs/DataJobServiceImpl.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/datajobs/DataJobServiceImpl.java
index 7db6c5c272..56ed6e30da 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/datajobs/DataJobServiceImpl.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/datajobs/DataJobServiceImpl.java
@@ -20,15 +20,27 @@
 
 package org.onap.cps.ncmp.impl.datajobs;
 
+import java.util.List;
+import java.util.Map;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.onap.cps.ncmp.api.datajobs.DataJobService;
 import org.onap.cps.ncmp.api.datajobs.models.DataJobMetadata;
 import org.onap.cps.ncmp.api.datajobs.models.DataJobReadRequest;
 import org.onap.cps.ncmp.api.datajobs.models.DataJobWriteRequest;
+import org.onap.cps.ncmp.api.datajobs.models.DmiWriteOperation;
+import org.onap.cps.ncmp.api.datajobs.models.ProducerKey;
+import org.onap.cps.ncmp.api.datajobs.models.SubJobWriteResponse;
+import org.springframework.stereotype.Service;
 
 @Slf4j
+@Service
+@RequiredArgsConstructor
 public class DataJobServiceImpl implements DataJobService {
 
+    private final DmiSubJobRequestHandler dmiSubJobClient;
+    private final WriteRequestExaminer writeRequestExaminer;
+
     @Override
     public void readDataJob(final String dataJobId, final DataJobMetadata dataJobMetadata,
                             final DataJobReadRequest dataJobReadRequest) {
@@ -36,8 +48,13 @@ public class DataJobServiceImpl implements DataJobService {
     }
 
     @Override
-    public void writeDataJob(final String dataJobId, final DataJobMetadata dataJobMetadata,
-                             final DataJobWriteRequest dataJobWriteRequest) {
+    public List<SubJobWriteResponse> writeDataJob(final String dataJobId, final DataJobMetadata dataJobMetadata,
+                                                  final DataJobWriteRequest dataJobWriteRequest) {
         log.info("data job id for write operation is: {}", dataJobId);
+
+        final Map<ProducerKey, List<DmiWriteOperation>> dmiWriteOperationsPerProducerKey =
+                writeRequestExaminer.splitDmiWriteOperationsFromRequest(dataJobId, dataJobWriteRequest);
+
+        return dmiSubJobClient.sendRequestsToDmi(dataJobId, dataJobMetadata, dmiWriteOperationsPerProducerKey);
     }
 }
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/datajobs/DmiSubJobRequestHandler.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/datajobs/DmiSubJobRequestHandler.java
new file mode 100644
index 0000000000..69eadab26a
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/datajobs/DmiSubJobRequestHandler.java
@@ -0,0 +1,85 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2024 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.impl.datajobs;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.onap.cps.ncmp.api.datajobs.models.DataJobMetadata;
+import org.onap.cps.ncmp.api.datajobs.models.DmiWriteOperation;
+import org.onap.cps.ncmp.api.datajobs.models.ProducerKey;
+import org.onap.cps.ncmp.api.datajobs.models.SubJobWriteRequest;
+import org.onap.cps.ncmp.api.datajobs.models.SubJobWriteResponse;
+import org.onap.cps.ncmp.api.impl.client.DmiRestClient;
+import org.onap.cps.ncmp.api.impl.config.DmiProperties;
+import org.onap.cps.ncmp.api.impl.operations.OperationType;
+import org.onap.cps.ncmp.api.impl.operations.RequiredDmiService;
+import org.onap.cps.ncmp.api.impl.utils.DmiServiceUrlBuilder;
+import org.onap.cps.utils.JsonObjectMapper;
+import org.springframework.http.ResponseEntity;
+import org.springframework.stereotype.Service;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class DmiSubJobRequestHandler {
+
+    private final DmiRestClient dmiRestClient;
+    private final DmiProperties dmiProperties;
+    private final JsonObjectMapper jsonObjectMapper;
+    static final String NO_AUTH_HEADER = null;
+
+    /**
+     * Sends sub-job write requests to the DMI Plugin.
+     *
+     * @param dataJobId                        data ojb identifier
+     * @param dataJobMetadata                  the data job's metadata
+     * @param dmiWriteOperationsPerProducerKey a collection of write requests per producer key.
+     * @return a list of sub-job write responses
+     */
+    public List<SubJobWriteResponse> sendRequestsToDmi(final String dataJobId, final DataJobMetadata dataJobMetadata,
+                                     final Map<ProducerKey, List<DmiWriteOperation>> dmiWriteOperationsPerProducerKey) {
+        final List<SubJobWriteResponse> subJobWriteResponses = new ArrayList<>(dmiWriteOperationsPerProducerKey.size());
+        dmiWriteOperationsPerProducerKey.forEach((producerKey, dmi3ggpWriteOperations) -> {
+            final SubJobWriteRequest subJobWriteRequest = new SubJobWriteRequest(dataJobMetadata.dataAcceptType(),
+                    dataJobMetadata.dataContentType(), dataJobId, dmi3ggpWriteOperations);
+
+            final String dmiResourceUrl = getDmiResourceUrl(dataJobId, producerKey);
+            final ResponseEntity<Object> responseEntity = dmiRestClient.postOperationWithJsonData(
+                    RequiredDmiService.DATA,
+                    dmiResourceUrl,
+                    jsonObjectMapper.asJsonString(subJobWriteRequest),
+                    OperationType.CREATE,
+                    NO_AUTH_HEADER);
+            final SubJobWriteResponse subJobWriteResponse = (SubJobWriteResponse) responseEntity.getBody();
+            log.debug("Sub job write response: {}", subJobWriteResponse);
+            subJobWriteResponses.add(subJobWriteResponse);
+        });
+        return subJobWriteResponses;
+    }
+
+    private String getDmiResourceUrl(final String dataJobId, final ProducerKey producerKey) {
+        return DmiServiceUrlBuilder.newInstance().pathSegment("writeJob").variablePathSegment("requestId", dataJobId)
+                .build(producerKey.dmiServiceName(), dmiProperties.getDmiBasePath());
+    }
+}
\ No newline at end of file
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/datajobs/WriteRequestExaminer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/datajobs/WriteRequestExaminer.java
new file mode 100644
index 0000000000..1c3e476c8b
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/datajobs/WriteRequestExaminer.java
@@ -0,0 +1,109 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2024 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.impl.datajobs;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.onap.cps.ncmp.api.datajobs.models.DataJobWriteRequest;
+import org.onap.cps.ncmp.api.datajobs.models.DmiWriteOperation;
+import org.onap.cps.ncmp.api.datajobs.models.ProducerKey;
+import org.onap.cps.ncmp.api.datajobs.models.WriteOperation;
+import org.onap.cps.ncmp.api.impl.utils.YangDataConverter;
+import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle;
+import org.onap.cps.ncmp.utils.AlternateIdMatcher;
+import org.onap.cps.spi.model.DataNode;
+import org.springframework.stereotype.Service;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class WriteRequestExaminer {
+
+    private final AlternateIdMatcher alternateIdMatcher;
+    private static final String PATH_SEPARATOR = "/";
+
+    /**
+     * Splitting incoming data job write request into Dmi Write Operations by ProducerKey.
+     *
+     * @param dataJobId data job identifier
+     * @param dataJobWriteRequest incoming data job write request
+     * @return {@code Map} map of Dmi Write Operations per Producer Key
+     */
+    public Map<ProducerKey, List<DmiWriteOperation>> splitDmiWriteOperationsFromRequest(
+            final String dataJobId,
+            final DataJobWriteRequest dataJobWriteRequest) {
+        final Map<ProducerKey, List<DmiWriteOperation>> dmiWriteOperationsPerProducerKey = new HashMap<>();
+        for (final WriteOperation writeOperation : dataJobWriteRequest.data()) {
+            examineWriteOperation(dataJobId, dmiWriteOperationsPerProducerKey, writeOperation);
+        }
+        return dmiWriteOperationsPerProducerKey;
+    }
+
+    private void examineWriteOperation(final String dataJobId,
+                                       final Map<ProducerKey, List<DmiWriteOperation>> dmiWriteOperationsPerProducerKey,
+                                       final WriteOperation writeOperation) {
+        log.debug("data job id for write operation is: {}", dataJobId);
+        final DataNode dataNode = alternateIdMatcher
+                .getCmHandleDataNodeByLongestMatchingAlternateId(writeOperation.path(), PATH_SEPARATOR);
+
+        final DmiWriteOperation dmiWriteOperation = createDmiWriteOperation(writeOperation, dataNode);
+
+        final ProducerKey producerKey = createProducerKey(dataNode);
+        final List<DmiWriteOperation> dmiWriteOperations;
+        if (dmiWriteOperationsPerProducerKey.containsKey(producerKey)) {
+            dmiWriteOperations = dmiWriteOperationsPerProducerKey.get(producerKey);
+        } else {
+            dmiWriteOperations = new ArrayList<>();
+            dmiWriteOperationsPerProducerKey.put(producerKey, dmiWriteOperations);
+        }
+        dmiWriteOperations.add(dmiWriteOperation);
+    }
+
+    private ProducerKey createProducerKey(final DataNode dataNode) {
+        return new ProducerKey((String) dataNode.getLeaves().get("dmi-service-name"),
+                (String) dataNode.getLeaves().get("data-producer-identifier"));
+    }
+
+    private DmiWriteOperation createDmiWriteOperation(final WriteOperation writeOperation,
+                                                      final DataNode dataNode) {
+        return new DmiWriteOperation(
+                writeOperation.path(),
+                writeOperation.op(),
+                (String) dataNode.getLeaves().get("module-set-tag"),
+                writeOperation.value(),
+                writeOperation.operationId(),
+                getPrivatePropertiesFromDataNode(dataNode));
+    }
+
+    private Map<String, String> getPrivatePropertiesFromDataNode(final DataNode dataNode) {
+        final YangModelCmHandle yangModelCmHandle = YangDataConverter.convertCmHandleToYangModel(dataNode);
+        final Map<String, String> cmHandleDmiProperties = new LinkedHashMap<>();
+        yangModelCmHandle.getDmiProperties()
+                .forEach(dmiProperty -> cmHandleDmiProperties.put(dmiProperty.getName(), dmiProperty.getValue()));
+        return cmHandleDmiProperties;
+    }
+
+}
\ No newline at end of file
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/utils/AlternateIdMatcher.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/utils/AlternateIdMatcher.java
index 8385f19f76..3fad684437 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/utils/AlternateIdMatcher.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/utils/AlternateIdMatcher.java
@@ -44,7 +44,7 @@ public class AlternateIdMatcher {
      * @param separator   a string that separates each element from the next.
      * @return data node
      */
-    public DataNode getCmHandleDataNodeByLongestMatchAlternateId(final String alternateId, final String separator) {
+    public DataNode getCmHandleDataNodeByLongestMatchingAlternateId(final String alternateId, final String separator) {
         String bestMatch = alternateId;
         while (StringUtils.isNotEmpty(bestMatch)) {
             try {
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/DataJobServiceImplSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/DataJobServiceImplSpec.groovy
index bef0adc9cb..9cee2bdbad 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/DataJobServiceImplSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/DataJobServiceImplSpec.groovy
@@ -25,6 +25,8 @@ import ch.qos.logback.classic.Logger
 import ch.qos.logback.classic.spi.ILoggingEvent
 import ch.qos.logback.core.read.ListAppender
 import org.onap.cps.ncmp.impl.datajobs.DataJobServiceImpl
+import org.onap.cps.ncmp.impl.datajobs.DmiSubJobRequestHandler
+import org.onap.cps.ncmp.impl.datajobs.WriteRequestExaminer
 import org.slf4j.LoggerFactory
 import org.onap.cps.ncmp.api.datajobs.models.DataJobReadRequest
 import org.onap.cps.ncmp.api.datajobs.models.DataJobWriteRequest
@@ -33,9 +35,14 @@ import org.onap.cps.ncmp.api.datajobs.models.ReadOperation
 import org.onap.cps.ncmp.api.datajobs.models.WriteOperation
 import spock.lang.Specification
 
-class DataJobServiceImplSpec extends Specification{
+class DataJobServiceImplSpec extends Specification {
 
-    def objectUnderTest = new DataJobServiceImpl()
+    def mockWriteRequestExaminer = Mock(WriteRequestExaminer)
+    def mockDmiSubJobRequestHandler = Mock(DmiSubJobRequestHandler)
+
+    def objectUnderTest = new DataJobServiceImpl(mockDmiSubJobRequestHandler, mockWriteRequestExaminer)
+
+    def myDataJobMetadata = new DataJobMetadata('', '', '')
 
     def logger = Spy(ListAppender<ILoggingEvent>)
 
@@ -47,28 +54,27 @@ class DataJobServiceImplSpec extends Specification{
         ((Logger) LoggerFactory.getLogger(DataJobServiceImpl.class)).detachAndStopAllAppenders()
     }
 
-    def '#operation data job request.'() {
-        given: 'data job metadata'
-            def dataJobMetadata = new DataJobMetadata('client-topic', 'application/vnd.3gpp.object-tree-hierarchical+json', 'application/3gpp-json-patch+json')
-        when: 'read/write data job request is processed'
-            if (operation == 'read') {
-                objectUnderTest.readDataJob('some-job-id', dataJobMetadata, new DataJobReadRequest([getWriteOrReadOperationRequest(operation)]))
-            } else {
-                objectUnderTest.writeDataJob('some-job-id', dataJobMetadata, new DataJobWriteRequest([getWriteOrReadOperationRequest(operation)]))
-            }
+    def 'Read data job request.'() {
+        when: 'read data job request is processed'
+            def readOperation = new ReadOperation('', '', '', [], [], '', '', 1)
+            objectUnderTest.readDataJob('my-job-id', myDataJobMetadata, new DataJobReadRequest([readOperation]))
         then: 'the data job id is correctly logged'
             def loggingEvent = logger.list[0]
             assert loggingEvent.level == Level.INFO
-            assert loggingEvent.formattedMessage.contains('data job id for ' + operation + ' operation is: some-job-id')
-        where: 'the following data job operations are used'
-            operation << ['read', 'write']
+            assert loggingEvent.formattedMessage.contains('data job id for read operation is: my-job-id')
     }
 
-    def getWriteOrReadOperationRequest(operation) {
-        if (operation == 'write') {
-            return new WriteOperation('some/write/path', 'add', 'some-operation-id', 'some-value')
-        }
-        return new ReadOperation('some/read/path', 'read', 'some-operation-id', ['some-attrib-1'], ['some-field-1'], 'some-filter', 'some-scope-type', 1)
+    def 'Write data-job request.'() {
+        given: 'data job metadata and write request'
+            def dataJobWriteRequest = new DataJobWriteRequest([new WriteOperation('', '', '', null)])
+        and: 'a map of producer key and dmi 3gpp write operation'
+            def dmiWriteOperationsPerProducerKey = [:]
+        when: 'write data job request is processed'
+            objectUnderTest.writeDataJob('my-job-id', myDataJobMetadata, dataJobWriteRequest)
+        then: 'the examiner service is called and a map is returned'
+            1 * mockWriteRequestExaminer.splitDmiWriteOperationsFromRequest('my-job-id', dataJobWriteRequest) >> dmiWriteOperationsPerProducerKey
+        and: 'the dmi request handler is called with the result from the examiner'
+            1 * mockDmiSubJobRequestHandler.sendRequestsToDmi('my-job-id', myDataJobMetadata, dmiWriteOperationsPerProducerKey)
     }
 
     def setupLogger() {
@@ -77,4 +83,4 @@ class DataJobServiceImplSpec extends Specification{
         setupLogger.addAppender(logger)
         logger.start()
     }
-}
+}
\ No newline at end of file
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/DmiSubJobRequestHandlerSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/DmiSubJobRequestHandlerSpec.groovy
new file mode 100644
index 0000000000..a1e0329792
--- /dev/null
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/DmiSubJobRequestHandlerSpec.groovy
@@ -0,0 +1,41 @@
+package org.onap.cps.ncmp.api.impl
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import org.onap.cps.ncmp.api.datajobs.models.DataJobMetadata
+import org.onap.cps.ncmp.api.datajobs.models.DmiWriteOperation
+import org.onap.cps.ncmp.api.datajobs.models.ProducerKey
+import org.onap.cps.ncmp.api.datajobs.models.SubJobWriteResponse
+import org.onap.cps.ncmp.api.impl.client.DmiRestClient
+import org.onap.cps.ncmp.api.impl.config.DmiProperties
+import org.onap.cps.ncmp.api.impl.operations.OperationType
+import org.onap.cps.ncmp.api.impl.operations.RequiredDmiService
+import org.onap.cps.ncmp.impl.datajobs.DmiSubJobRequestHandler
+import org.onap.cps.utils.JsonObjectMapper
+import org.springframework.http.HttpStatus
+import org.springframework.http.ResponseEntity
+import spock.lang.Specification
+
+class DmiSubJobRequestHandlerSpec extends Specification {
+
+    def mockDmiRestClient = Mock(DmiRestClient)
+    def jsonObjectMapper = new JsonObjectMapper(new ObjectMapper())
+    def mockDmiProperties = Mock(DmiProperties)
+    def static NO_AUTH = null
+    def objectUnderTest = new DmiSubJobRequestHandler(mockDmiRestClient, mockDmiProperties, jsonObjectMapper)
+
+    def 'Send a sub-job request to the DMI Plugin.'() {
+        given: 'a data job id, metadata and a map of producer keys and write operations to create a request'
+            def dataJobId = 'some-job-id'
+            def dataJobMetadata = new DataJobMetadata('', '', '')
+            def dmiWriteOperation = new DmiWriteOperation('', '', '', null, '', [:])
+            def dmiWriteOperationsPerProducerKey = [new ProducerKey('', ''): [dmiWriteOperation]]
+            def response = new ResponseEntity<>(new SubJobWriteResponse('my-sub-job-id', '', ''), HttpStatus.OK)
+        when: 'sending request to DMI invoked'
+            objectUnderTest.sendRequestsToDmi(dataJobId, dataJobMetadata, dmiWriteOperationsPerProducerKey)
+        then: 'the dmi rest client is called'
+            1 * mockDmiRestClient.postOperationWithJsonData(RequiredDmiService.DATA, _, _, OperationType.CREATE, NO_AUTH) >> response
+        and: 'the result contains the expected sub-job write responses'
+            def result = response.body
+            assert result.subJobId() == 'my-sub-job-id'
+    }
+}
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/WriteRequestExaminerSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/WriteRequestExaminerSpec.groovy
new file mode 100644
index 0000000000..d5d6339422
--- /dev/null
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/WriteRequestExaminerSpec.groovy
@@ -0,0 +1,63 @@
+
+package org.onap.cps.ncmp.impl.datajobs
+
+import org.onap.cps.ncmp.api.datajobs.models.DataJobWriteRequest
+import org.onap.cps.ncmp.api.datajobs.models.WriteOperation
+import org.onap.cps.ncmp.utils.AlternateIdMatcher
+import org.onap.cps.spi.model.DataNode
+import spock.lang.Specification
+
+class WriteRequestExaminerSpec extends Specification {
+
+    def mockAlternateIdMatcher = Mock(AlternateIdMatcher)
+    def objectUnderTest = new WriteRequestExaminer(mockAlternateIdMatcher)
+
+    def setup() {
+        def ch1 = new DataNode(leaves: [id: 'ch1', 'dmi-service-name': 'dmiA', 'data-producer-identifier': 'p1'])
+        def ch2 = new DataNode(leaves: [id: 'ch2', 'dmi-service-name': 'dmiA', 'data-producer-identifier': 'p1'])
+        def ch3 = new DataNode(leaves: [id: 'ch3', 'dmi-service-name': 'dmiA', 'data-producer-identifier': 'p2'])
+        def ch4 = new DataNode(leaves: [id: 'ch4', 'dmi-service-name': 'dmiB', 'data-producer-identifier': 'p1'])
+        mockAlternateIdMatcher.getCmHandleDataNodeByLongestMatchingAlternateId('fdn1', '/') >> ch1
+        mockAlternateIdMatcher.getCmHandleDataNodeByLongestMatchingAlternateId('fdn2', '/') >> ch2
+        mockAlternateIdMatcher.getCmHandleDataNodeByLongestMatchingAlternateId('fdn3', '/') >> ch3
+        mockAlternateIdMatcher.getCmHandleDataNodeByLongestMatchingAlternateId('fdn4', '/') >> ch4
+    }
+
+    def 'Create a map of dmi write requests per producer key with #scenario.'() {
+        given: 'a write request with some write operations'
+            def writeOperations = writeOperationFdns.collect {
+                new WriteOperation(it, '', '', null)
+            }
+        and: 'operations are wrapped in a write request'
+            def dataJobWriteRequest = new DataJobWriteRequest(writeOperations)
+        when: 'the DMI write operations are split from the request'
+            def dmiWriteOperationsPerProducerKey = objectUnderTest.splitDmiWriteOperationsFromRequest('some id', dataJobWriteRequest)
+        then: 'we get the expected number of keys and values.'
+            def producerKeysAsStrings = dmiWriteOperationsPerProducerKey.keySet().collect {
+                it.toString()
+            }
+            assert producerKeysAsStrings.size() == expectedKeys.size()
+            assert expectedKeys.containsAll(producerKeysAsStrings)
+        where:
+            scenario                                                          | writeOperationFdns               || expectedKeys
+            'one fdn'                                                         | ['fdn1']                         || ['dmiA#p1']
+            'a duplicated target'                                             | ['fdn1','fdn1']                  || ['dmiA#p1']
+            'two different targets'                                           | ['fdn1','fdn2']                  || ['dmiA#p1']
+            'two different targets and different producer keys'               | ['fdn1','fdn3']                  || ['dmiA#p1', 'dmiA#p2']
+            'two different targets and different DMI services'                | ['fdn1','fdn4']                  || ['dmiA#p1', 'dmiB#p1']
+            'many targets with different dmi service names and producer keys' | ['fdn1', 'fdn2', 'fdn3', 'fdn4'] || ['dmiA#p1', 'dmiA#p2', 'dmiB#p1']
+    }
+
+    def 'Validate the ordering of the created sub jobs.'() {
+        given: 'a few write operations for the same producer'
+            def writeOperations = (1..3).collect {
+                new WriteOperation('fdn1', '', it.toString(), null)
+            }
+        and: 'operation is wrapped in a write request'
+            def dataJobWriteRequest = new DataJobWriteRequest(writeOperations)
+        when: 'the DMI write operations are split from the request'
+            def dmiWriteOperations = objectUnderTest.splitDmiWriteOperationsFromRequest('some id', dataJobWriteRequest).values().iterator().next()
+        then: 'we get the operation ids in the expected order.'
+            assert dmiWriteOperations.operationId == ['1', '2', '3']
+    }
+}
\ No newline at end of file
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/utils/AlternateIdMatcherSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/utils/AlternateIdMatcherSpec.groovy
index 720a7e7e9f..3422610450 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/utils/AlternateIdMatcherSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/utils/AlternateIdMatcherSpec.groovy
@@ -40,7 +40,7 @@ class AlternateIdMatcherSpec extends Specification {
 
     def 'Finding longest alternate id matches.'() {
         expect: 'querying for alternate id a matching result found'
-            assert objectUnderTest.getCmHandleDataNodeByLongestMatchAlternateId(targetAlternateId, '/') != null
+            assert objectUnderTest.getCmHandleDataNodeByLongestMatchingAlternateId(targetAlternateId, '/') != null
         where: 'the following parameters are used'
             scenario                   | targetAlternateId
             'exact match'              | '/a/b'
@@ -51,7 +51,7 @@ class AlternateIdMatcherSpec extends Specification {
 
     def 'Attempt to find longest alternate id match without any matches.'() {
         when: 'attempt to find alternateId'
-            objectUnderTest.getCmHandleDataNodeByLongestMatchAlternateId(targetAlternateId, '/')
+            objectUnderTest.getCmHandleDataNodeByLongestMatchingAlternateId(targetAlternateId, '/')
         then: 'no alternate id match found exception thrown'
             def thrown = thrown(NoAlternateIdMatchFoundException)
         and: 'the exception has the relevant details from the error response'