Async: NCMP Rest impl. including Request ID generation

- Restructured code and moved some of them at controller and service
  layer.
- Unit is fixed and organized to it's belonging classes.

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

7 files changed, 161 insertions, 137 deletions

diff --git a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/NetworkCmProxyController.java b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/NetworkCmProxyController.java
index de6c3c4f3d..19b9193330 100755
--- a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/NetworkCmProxyController.java
+++ b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/NetworkCmProxyController.java
@@ -31,13 +31,17 @@ import static org.onap.cps.ncmp.api.impl.operations.DmiRequestBody.OperationEnum
 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.UUID;
 import java.util.stream.Collectors;
 import javax.validation.Valid;
 import javax.validation.constraints.NotNull;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.onap.cps.ncmp.api.NetworkCmProxyDataService;
+import org.onap.cps.ncmp.api.impl.exception.InvalidTopicException;
 import org.onap.cps.ncmp.api.models.NcmpServiceCmHandle;
 import org.onap.cps.ncmp.rest.api.NetworkCmProxyApi;
 import org.onap.cps.ncmp.rest.model.CmHandleProperties;
@@ -50,6 +54,7 @@ import org.onap.cps.ncmp.rest.model.ModuleNameAsJsonObject;
 import org.onap.cps.ncmp.rest.model.ModuleNamesAsJsonArray;
 import org.onap.cps.ncmp.rest.model.RestModuleReference;
 import org.onap.cps.ncmp.rest.model.RestOutputCmHandle;
+import org.onap.cps.utils.CpsValidator;
 import org.onap.cps.utils.JsonObjectMapper;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
@@ -63,6 +68,9 @@ import org.springframework.web.bind.annotation.RestController;
 public class NetworkCmProxyController implements NetworkCmProxyApi {
 
     private static final String NO_BODY = null;
+    private static final String NO_REQUEST_ID = null;
+    private static final String NO_TOPIC = null;
+    public static final String ASYNC_REQUEST_ID = "requestId";
 
     private final NetworkCmProxyDataService networkCmProxyDataService;
     private final JsonObjectMapper jsonObjectMapper;
@@ -82,11 +90,19 @@ public class NetworkCmProxyController implements NetworkCmProxyApi {
                                                                         final @NotNull @Valid String resourceIdentifier,
                                                                         final @Valid String optionsParamInQuery,
                                                                         final @Valid String topicParamInQuery) {
+        final ResponseEntity<Map<String, Object>> asyncResponse = populateAsyncResponse(topicParamInQuery);
+        final Map<String, Object> asyncResponseData = asyncResponse.getBody();
+
         final Object responseObject = networkCmProxyDataService.getResourceDataOperationalForCmHandle(cmHandle,
                 resourceIdentifier,
                 optionsParamInQuery,
-                topicParamInQuery);
-        return ResponseEntity.ok(responseObject);
+                asyncResponseData == null ? NO_TOPIC : topicParamInQuery,
+                asyncResponseData == null ? NO_REQUEST_ID : asyncResponseData.get(ASYNC_REQUEST_ID).toString());
+
+        if (asyncResponseData == null) {
+            return ResponseEntity.ok(responseObject);
+        }
+        return ResponseEntity.ok(asyncResponse);
     }
 
     /**
@@ -103,11 +119,19 @@ public class NetworkCmProxyController implements NetworkCmProxyApi {
                                                                     final @NotNull @Valid String resourceIdentifier,
                                                                     final @Valid String optionsParamInQuery,
                                                                     final @Valid String topicParamInQuery) {
+        final ResponseEntity<Map<String, Object>> asyncResponse = populateAsyncResponse(topicParamInQuery);
+        final Map<String, Object> asyncResponseData = asyncResponse.getBody();
+
         final Object responseObject = networkCmProxyDataService.getResourceDataPassThroughRunningForCmHandle(cmHandle,
                 resourceIdentifier,
                 optionsParamInQuery,
-                topicParamInQuery);
-        return ResponseEntity.ok(responseObject);
+                asyncResponseData == null ? NO_TOPIC : topicParamInQuery,
+                asyncResponseData == null ? NO_REQUEST_ID : asyncResponseData.get(ASYNC_REQUEST_ID).toString());
+
+        if (asyncResponseData == null) {
+            return ResponseEntity.ok(responseObject);
+        }
+        return ResponseEntity.ok(asyncResponse);
     }
 
     @Override
@@ -257,4 +281,33 @@ public class NetworkCmProxyController implements NetworkCmProxyApi {
         restOutputCmHandle.setPublicCmHandleProperties(cmHandlePublicProperties);
         return restOutputCmHandle;
     }
+
+    private ResponseEntity<Map<String, Object>> populateAsyncResponse(final String topicParamInQuery) {
+        final boolean processAsynchronously = hasTopicParameter(topicParamInQuery);
+        final Map<String, Object> responseData;
+        if (processAsynchronously) {
+            responseData = getAsyncResponseData();
+        } else {
+            responseData = null;
+        }
+        return ResponseEntity.ok().body(responseData);
+    }
+
+    private static boolean hasTopicParameter(final String topicName) {
+        if (topicName == null) {
+            return false;
+        }
+        if (CpsValidator.validateTopicName(topicName)) {
+            return true;
+        }
+        throw new InvalidTopicException("Topic name " + topicName + " is invalid", "invalid topic");
+    }
+
+    private Map<String, Object> getAsyncResponseData() {
+        final Map<String, Object> asyncResponseData = new HashMap<>(1);
+        final String resourceDataRequestId = UUID.randomUUID().toString();
+        asyncResponseData.put(ASYNC_REQUEST_ID, resourceDataRequestId);
+        return asyncResponseData;
+    }
+
 }
diff --git a/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/controller/NetworkCmProxyControllerSpec.groovy b/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/controller/NetworkCmProxyControllerSpec.groovy
index 88633450b3..d79d9622e3 100644
--- a/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/controller/NetworkCmProxyControllerSpec.groovy
+++ b/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/controller/NetworkCmProxyControllerSpec.groovy
@@ -72,6 +72,7 @@ class NetworkCmProxyControllerSpec extends Specification {
 
     @Shared
     def NO_TOPIC = null
+    def NO_REQUEST_ID = null
 
     def 'Get Resource Data from pass-through operational.'() {
         given: 'resource data url'
@@ -86,14 +87,15 @@ class NetworkCmProxyControllerSpec extends Specification {
             1 * mockNetworkCmProxyDataService.getResourceDataOperationalForCmHandle('testCmHandle',
                     'parent/child',
                     '(a=1,b=2)',
-                    NO_TOPIC)
+                    NO_TOPIC,
+                    NO_REQUEST_ID)
         and: 'response status is Ok'
             response.status == HttpStatus.OK.value()
     }
 
-    def 'Get Resource Data from pass-through operational with #scenario.'() {
+    def 'Get Resource Data from #datastoreInUrl with #scenario.'() {
         given: 'resource data url'
-            def getUrl = "$ncmpBasePathV1/ch/testCmHandle/data/ds/ncmp-datastore:passthrough-operational" +
+            def getUrl = "$ncmpBasePathV1/ch/testCmHandle/data/ds/ncmp-datastore:${datastoreInUrl}" +
                     "?resourceIdentifier=parent/child&options=(a=1,b=2)${topicQueryParam}"
         when: 'get data resource request is performed'
             def response = mvc.perform(
@@ -101,19 +103,30 @@ class NetworkCmProxyControllerSpec extends Specification {
                     .contentType(MediaType.APPLICATION_JSON)
             ).andReturn().response
         then: 'the NCMP data service is called with operational data for cm handle'
-            1 * mockNetworkCmProxyDataService.getResourceDataOperationalForCmHandle('testCmHandle',
+            expectedNumberOfMethodExecutions
+                    * mockNetworkCmProxyDataService."${expectedMethodName}"('testCmHandle',
                     'parent/child',
                     '(a=1,b=2)',
-                    expectedTopicName)
-        and: 'response status is Ok'
-            response.status == HttpStatus.OK.value()
+                    expectedTopicName,
+                    _)
+        then: 'response status is expected'
+            response.status == expectedHttpStatus
         where: 'the following parameters are used'
-            scenario               | topicQueryParam        || expectedTopicName
-            'Url with valid topic' | "&topic=my-topic-name" || "my-topic-name"
-            'No topic in url'      | ''                     || NO_TOPIC
-            'Null topic in url'    | "&topic=null"          || "null"
-            'Empty topic in url'   | "&topic=\"\""          || "\"\""
-            'Missing topic in url' | "&topic="              || ""
+            scenario                               | datastoreInUrl            | topicQueryParam        || expectedTopicName | expectedMethodName                             | expectedNumberOfMethodExecutions | expectedHttpStatus
+            'url with valid topic'                 | 'passthrough-operational' | '&topic=my-topic-name' || 'my-topic-name'   | 'getResourceDataOperationalForCmHandle'        | 1                                | HttpStatus.OK.value()
+            'no topic in url'                      | 'passthrough-operational' | ''                     || NO_TOPIC          | 'getResourceDataOperationalForCmHandle'        | 1                                | HttpStatus.OK.value()
+            'null topic in url'                    | 'passthrough-operational' | '&topic=null'          || 'null'            | 'getResourceDataOperationalForCmHandle'        | 1                                | HttpStatus.OK.value()
+            'empty topic in url'                   | 'passthrough-operational' | '&topic=\"\"'          || null              | 'getResourceDataOperationalForCmHandle'        | 0                                | HttpStatus.BAD_REQUEST.value()
+            'missing topic in url'                 | 'passthrough-operational' | '&topic='              || null              | 'getResourceDataOperationalForCmHandle'        | 0                                | HttpStatus.BAD_REQUEST.value()
+            'blank topic value in url'             | 'passthrough-operational' | '&topic=\" \"'         || null              | 'getResourceDataOperationalForCmHandle'        | 0                                | HttpStatus.BAD_REQUEST.value()
+            'invalid non-empty topic value in url' | 'passthrough-operational' | '&topic=1_5_*_#'       || null              | 'getResourceDataOperationalForCmHandle'        | 0                                | HttpStatus.BAD_REQUEST.value()
+            'url with valid topic'                 | 'passthrough-running'     | '&topic=my-topic-name' || 'my-topic-name'   | 'getResourceDataPassThroughRunningForCmHandle' | 1                                | HttpStatus.OK.value()
+            'no topic in url'                      | 'passthrough-running'     | ''                     || NO_TOPIC          | 'getResourceDataPassThroughRunningForCmHandle' | 1                                | HttpStatus.OK.value()
+            'null topic in url'                    | 'passthrough-running'     | '&topic=null'          || 'null'            | 'getResourceDataPassThroughRunningForCmHandle' | 1                                | HttpStatus.OK.value()
+            'empty topic in url'                   | 'passthrough-running'     | '&topic=\"\"'          || null              | 'getResourceDataPassThroughRunningForCmHandle' | 0                                | HttpStatus.BAD_REQUEST.value()
+            'missing topic in url'                 | 'passthrough-running'     | '&topic='              || null              | 'getResourceDataPassThroughRunningForCmHandle' | 0                                | HttpStatus.BAD_REQUEST.value()
+            'blank topic value in url'             | 'passthrough-running'     | '&topic=\" \"'         || null              | 'getResourceDataPassThroughRunningForCmHandle' | 0                                | HttpStatus.BAD_REQUEST.value()
+            'invalid non-empty topic value in url' | 'passthrough-running'     | '&topic=1_5_*_#'       || null              | 'getResourceDataPassThroughRunningForCmHandle' | 0                                | HttpStatus.BAD_REQUEST.value()
     }
 
     def 'Get Resource Data from pass-through running with #scenario value in resource identifier param.'() {
@@ -124,7 +137,8 @@ class NetworkCmProxyControllerSpec extends Specification {
             mockNetworkCmProxyDataService.getResourceDataPassThroughRunningForCmHandle('testCmHandle',
                     resourceIdentifier,
                     '(a=1,b=2)',
-                    NO_TOPIC) >> '{valid-json}'
+                    NO_TOPIC,
+                    NO_REQUEST_ID) >> '{valid-json}'
         when: 'get data resource request is performed'
             def response = mvc.perform(
                     get(getUrl)
@@ -271,5 +285,24 @@ class NetworkCmProxyControllerSpec extends Specification {
         and: 'the response is No Content'
             response.status == HttpStatus.NO_CONTENT.value()
     }
+
+    def 'Get resource data from DMI with valid topic i.e. async request for #scenario'() {
+        given: 'resource data url'
+            def getUrl = "$ncmpBasePathV1/ch/testCmHandle/data/ds/ncmp-datastore:${datastoreInUrl}" +
+                    "?resourceIdentifier=parent/child&options=(a=1,b=2)&topic=my-topic-name"
+        when: 'get data resource request is performed'
+            def response = mvc.perform(
+                    get(getUrl)
+                            .contentType(MediaType.APPLICATION_JSON)
+                            .accept(MediaType.APPLICATION_JSON_VALUE)
+            ).andReturn().response
+        then: 'async request id is generated'
+            assert response.contentAsString.contains("requestId")
+        where: 'the following parameters are used'
+            scenario                   | datastoreInUrl
+            ':passthrough-operational' | 'passthrough-operational'
+            ':passthrough-running'     | 'passthrough-running'
+    }
+
 }
 
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/NetworkCmProxyDataService.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/NetworkCmProxyDataService.java
index d50b8c5ea8..cbd0756c09 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/NetworkCmProxyDataService.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/NetworkCmProxyDataService.java
@@ -52,12 +52,14 @@ public interface NetworkCmProxyDataService {
      * @param resourceIdentifier resource identifier
      * @param optionsParamInQuery options query
      * @param topicParamInQuery topic name for (triggering) async responses
+     * @param requestId unique requestId for async request
      * @return {@code Object} resource data
      */
     Object getResourceDataOperationalForCmHandle(String cmHandleId,
                                                  String resourceIdentifier,
                                                  String optionsParamInQuery,
-                                                 String topicParamInQuery);
+                                                 String topicParamInQuery,
+                                                 String requestId);
 
     /**
      * Get resource data for data store pass-through running
@@ -66,13 +68,15 @@ public interface NetworkCmProxyDataService {
      * @param cmHandleId cm handle identifier
      * @param resourceIdentifier resource identifier
      * @param optionsParamInQuery options query
-     * @param topicParamInQuery topic query
+     * @param topicParamInQuery topic name for (triggering) async responses
+     * @param requestId unique requestId for async request
      * @return {@code Object} resource data
      */
     Object getResourceDataPassThroughRunningForCmHandle(String cmHandleId,
                                                         String resourceIdentifier,
                                                         String optionsParamInQuery,
-                                                        String topicParamInQuery);
+                                                        String topicParamInQuery,
+                                                        String requestId);
 
     /**
      * Write resource data for data store pass-through running
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/NetworkCmProxyDataServiceImpl.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/NetworkCmProxyDataServiceImpl.java
index 81c060eb3b..696bb0a37c 100755
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/NetworkCmProxyDataServiceImpl.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/NetworkCmProxyDataServiceImpl.java
@@ -37,8 +37,6 @@ import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.UUID;
-import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -47,7 +45,6 @@ import org.onap.cps.api.CpsDataService;
 import org.onap.cps.api.CpsModuleService;
 import org.onap.cps.ncmp.api.NetworkCmProxyDataService;
 import org.onap.cps.ncmp.api.impl.exception.HttpClientRequestException;
-import org.onap.cps.ncmp.api.impl.exception.InvalidTopicException;
 import org.onap.cps.ncmp.api.impl.operations.DmiDataOperations;
 import org.onap.cps.ncmp.api.impl.operations.DmiModelOperations;
 import org.onap.cps.ncmp.api.impl.operations.DmiOperations;
@@ -65,7 +62,6 @@ import org.onap.cps.spi.exceptions.SchemaSetNotFoundException;
 import org.onap.cps.spi.model.ModuleReference;
 import org.onap.cps.utils.CpsValidator;
 import org.onap.cps.utils.JsonObjectMapper;
-import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Service;
 
@@ -90,12 +86,6 @@ public class NetworkCmProxyDataServiceImpl implements NetworkCmProxyDataService
 
     private final YangModelCmHandleRetriever yangModelCmHandleRetriever;
 
-    // valid kafka topic name regex
-    private static final Pattern TOPIC_NAME_PATTERN = Pattern.compile("^[a-zA-Z0-9]([._-](?![._-])|"
-            + "[a-zA-Z0-9]){0,120}[a-zA-Z0-9]$");
-    private static final String NO_REQUEST_ID = null;
-    private static final String NO_TOPIC = null;
-
     @Override
     public DmiPluginRegistrationResponse updateDmiRegistrationAndSyncModule(
         final DmiPluginRegistration dmiPluginRegistration) {
@@ -119,20 +109,22 @@ public class NetworkCmProxyDataServiceImpl implements NetworkCmProxyDataService
     public Object getResourceDataOperationalForCmHandle(final String cmHandleId,
                                                         final String resourceIdentifier,
                                                         final String optionsParamInQuery,
-                                                        final String topicParamInQuery) {
+                                                        final String topicParamInQuery,
+                                                        final String requestId) {
         CpsValidator.validateNameCharacters(cmHandleId);
-        return validateTopicNameAndGetResourceData(cmHandleId, resourceIdentifier,
-                DmiOperations.DataStoreEnum.PASSTHROUGH_OPERATIONAL, optionsParamInQuery, topicParamInQuery);
+        return getResourceDataResponse(cmHandleId, resourceIdentifier,
+                DmiOperations.DataStoreEnum.PASSTHROUGH_OPERATIONAL, optionsParamInQuery, topicParamInQuery, requestId);
     }
 
     @Override
     public Object getResourceDataPassThroughRunningForCmHandle(final String cmHandleId,
                                                                final String resourceIdentifier,
                                                                final String optionsParamInQuery,
-                                                               final String topicParamInQuery) {
+                                                               final String topicParamInQuery,
+                                                               final String requestId) {
         CpsValidator.validateNameCharacters(cmHandleId);
-        return validateTopicNameAndGetResourceData(cmHandleId, resourceIdentifier,
-                DmiOperations.DataStoreEnum.PASSTHROUGH_RUNNING, optionsParamInQuery, topicParamInQuery);
+        return getResourceDataResponse(cmHandleId, resourceIdentifier,
+                DmiOperations.DataStoreEnum.PASSTHROUGH_RUNNING, optionsParamInQuery, topicParamInQuery, requestId);
     }
 
     @Override
@@ -329,35 +321,14 @@ public class NetworkCmProxyDataServiceImpl implements NetworkCmProxyDataService
             yangModelCmHandle.getId());
     }
 
-    private static boolean hasTopicParameter(final String topicName) {
-        if (topicName == null) {
-            return false;
-        }
-        if (TOPIC_NAME_PATTERN.matcher(topicName).matches()) {
-            return true;
-        }
-        throw new InvalidTopicException("Topic name " + topicName + " is invalid", "invalid topic");
-    }
-
-    private Map<String, Object> buildDmiResponse(final String requestId) {
-        final Map<String, Object> dmiResponseMap = new HashMap<>();
-        dmiResponseMap.put("requestId", requestId);
-        return dmiResponseMap;
-    }
-
-    private Object validateTopicNameAndGetResourceData(final String cmHandleId,
-                                                       final String resourceIdentifier,
-                                                       final DmiOperations.DataStoreEnum dataStore,
-                                                       final String optionsParamInQuery,
-                                                       final String topicParamInQuery) {
-        final boolean processAsynchronously = hasTopicParameter(topicParamInQuery);
-        if (processAsynchronously) {
-            final String resourceDataRequestId = UUID.randomUUID().toString();
-            return ResponseEntity.status(HttpStatus.OK)
-                    .body(buildDmiResponse(resourceDataRequestId));
-        }
+    private Object getResourceDataResponse(final String cmHandleId,
+                                           final String resourceIdentifier,
+                                           final DmiOperations.DataStoreEnum dataStore,
+                                           final String optionsParamInQuery,
+                                           final String topicParamInQuery,
+                                           final String requestId) {
         final ResponseEntity<?> responseEntity = dmiDataOperations.getResourceDataFromDmi(
-                cmHandleId, resourceIdentifier, optionsParamInQuery, dataStore, NO_REQUEST_ID, NO_TOPIC);
+                cmHandleId, resourceIdentifier, optionsParamInQuery, dataStore, requestId, topicParamInQuery);
         return handleResponse(responseEntity, OperationEnum.READ);
     }
 }
\ No newline at end of file
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/NetworkCmProxyDataServiceImplSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/NetworkCmProxyDataServiceImplSpec.groovy
index 489c71c0e9..2d01dba522 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/NetworkCmProxyDataServiceImplSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/NetworkCmProxyDataServiceImplSpec.groovy
@@ -23,7 +23,6 @@
 package org.onap.cps.ncmp.api.impl
 
 import org.onap.cps.ncmp.api.impl.exception.HttpClientRequestException
-import org.onap.cps.ncmp.api.impl.exception.InvalidTopicException
 import org.onap.cps.ncmp.api.impl.operations.YangModelCmHandleRetriever
 import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle
 import spock.lang.Shared
@@ -119,7 +118,8 @@ class NetworkCmProxyDataServiceImplSpec extends Specification {
             def response = objectUnderTest.getResourceDataOperationalForCmHandle('testCmHandle',
                     'testResourceId',
                     OPTIONS_PARAM,
-                    NO_TOPIC)
+                    NO_TOPIC,
+                    NO_REQUEST_ID)
         then: 'DMI returns a json response'
             response == 'dmi-response'
     }
@@ -137,7 +137,8 @@ class NetworkCmProxyDataServiceImplSpec extends Specification {
             objectUnderTest.getResourceDataOperationalForCmHandle('testCmHandle',
                     'testResourceId',
                     OPTIONS_PARAM,
-                    NO_TOPIC)
+                    NO_TOPIC,
+                    NO_REQUEST_ID)
         then: 'exception is thrown with the expected response code and details'
             def exceptionThrown = thrown(HttpClientRequestException.class)
             exceptionThrown.details.contains('NOK-json')
@@ -160,7 +161,8 @@ class NetworkCmProxyDataServiceImplSpec extends Specification {
             objectUnderTest.getResourceDataOperationalForCmHandle('testCmHandle',
                     'testResourceId',
                     OPTIONS_PARAM,
-                    NO_TOPIC)
+                    NO_TOPIC,
+                    NO_REQUEST_ID)
         then: 'exception is thrown'
             def exceptionThrown = thrown(HttpClientRequestException.class)
         and: 'details contain the original response'
@@ -183,7 +185,8 @@ class NetworkCmProxyDataServiceImplSpec extends Specification {
             def response = objectUnderTest.getResourceDataPassThroughRunningForCmHandle('testCmHandle',
                     'testResourceId',
                     OPTIONS_PARAM,
-                    NO_TOPIC)
+                    NO_TOPIC,
+                    NO_REQUEST_ID)
         then: 'get resource data returns expected response'
             response == '{dmi-response}'
     }
@@ -204,7 +207,8 @@ class NetworkCmProxyDataServiceImplSpec extends Specification {
             objectUnderTest.getResourceDataPassThroughRunningForCmHandle('testCmHandle',
                     'testResourceId',
                     OPTIONS_PARAM,
-                    NO_TOPIC)
+                    NO_TOPIC,
+                    NO_REQUEST_ID)
         then: 'exception is thrown'
             def exceptionThrown = thrown(HttpClientRequestException.class)
         and: 'details contain the original response'
@@ -212,72 +216,6 @@ class NetworkCmProxyDataServiceImplSpec extends Specification {
             exceptionThrown.httpStatus == HttpStatus.NOT_FOUND.value()
     }
 
-    def 'DMI Operational data request with #scenario'() {
-        given: 'cps data service returns valid data node'
-            mockCpsDataService.getDataNode('NCMP-Admin', 'ncmp-dmi-registry',
-                    cmHandleXPath, FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS) >> dataNode
-        and: 'dmi data operation returns valid response and data'
-            mockDmiDataOperations.getResourceDataFromDmi(_, _, _, _, _, NO_REQUEST_ID, NO_TOPIC)
-                    >> new ResponseEntity<>('{dmi-response}', HttpStatus.OK)
-        when: 'get resource data is called data operational with blank topic'
-            def responseData = objectUnderTest.getResourceDataOperationalForCmHandle('', '',
-                    '', emptyTopic)
-        then: 'a invalid topic exception is thrown'
-            thrown(InvalidTopicException)
-        where: 'the following parameters are used'
-            scenario                               | emptyTopic
-            'no topic value in url'                | ''
-            'empty topic value in url'             | '\"\"'
-            'blank topic value in url'             | ' '
-            'invalid non-empty topic value in url' | '1_5_*_#'
-    }
-
-    def 'Get resource data for data operational from DMI with valid topic i.e. async request.'() {
-        given: 'cps data service returns valid data node'
-            mockCpsDataService.getDataNode(*_) >> dataNode
-        and: 'dmi data operation returns valid response and data'
-            mockDmiDataOperations.getResourceDataFromDmi(_, _, _, _, _, 'my-topic-name')
-                    >> new ResponseEntity<>('{dmi-response}', HttpStatus.OK)
-        when: 'get resource data is called for data operational with valid topic'
-            def responseData = objectUnderTest.getResourceDataOperationalForCmHandle('', '', '', 'my-topic-name')
-        then: 'non empty request id is generated'
-            assert responseData.body.requestId.length() > 0
-    }
-
-    def 'Get resource data for pass through running from DMI with valid topic async request.'() {
-        given: 'cps data service returns valid data node'
-            mockCpsDataService.getDataNode('NCMP-Admin', 'ncmp-dmi-registry',
-                    cmHandleXPath, FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS) >> dataNode
-        and: 'dmi data operation returns valid response and data'
-            mockDmiDataOperations.getResourceDataFromDmi(_, _, _, _, _, 'my-topic-name')
-                    >> new ResponseEntity<>('{dmi-response}', HttpStatus.OK)
-        when: 'get resource data is called for data operational with valid topic'
-            def responseData = objectUnderTest.getResourceDataPassThroughRunningForCmHandle('',
-                    '', OPTIONS_PARAM, 'my-topic-name')
-        then: 'non empty request id is generated'
-            assert responseData.body.requestId.length() > 0
-    }
-
-    def 'DMI pass through running data request with #scenario'() {
-        given: 'cps data service returns valid data node'
-            mockCpsDataService.getDataNode('NCMP-Admin', 'ncmp-dmi-registry',
-                    cmHandleXPath, FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS) >> dataNode
-        and: 'dmi data operation returns valid response and data'
-            mockDmiDataOperations.getResourceDataFromDmi(_, _, _, _, NO_REQUEST_ID, NO_TOPIC)
-                    >> new ResponseEntity<>('{dmi-response}', HttpStatus.OK)
-        when: 'get resource data is called for data operational with valid topic'
-            def responseData = objectUnderTest.getResourceDataPassThroughRunningForCmHandle('',
-                    '', '', emptyTopic)
-        then: 'a invalid topic exception is thrown'
-            thrown(InvalidTopicException)
-        where: 'the following parameters are used'
-            scenario                               | emptyTopic
-            'no topic value in url'                | ''
-            'empty topic value in url'             | '\"\"'
-            'blank topic value in url'             | ' '
-            'invalid non-empty topic value in url' | '1_5_*_#'
-    }
-
     def 'Getting Yang Resources.'() {
         when: 'yang resources is called'
             objectUnderTest.getYangResourcesModuleReferences('some-cm-handle')
diff --git a/cps-service/src/main/java/org/onap/cps/utils/CpsValidator.java b/cps-service/src/main/java/org/onap/cps/utils/CpsValidator.java
index dd16495638..28b49c9666 100644
--- a/cps-service/src/main/java/org/onap/cps/utils/CpsValidator.java
+++ b/cps-service/src/main/java/org/onap/cps/utils/CpsValidator.java
@@ -22,6 +22,7 @@ package org.onap.cps.utils;
 
 import com.google.common.collect.Lists;
 import java.util.Collection;
+import java.util.regex.Pattern;
 import lombok.AccessLevel;
 import lombok.NoArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -32,6 +33,8 @@ import org.onap.cps.spi.exceptions.DataValidationException;
 public final class CpsValidator {
 
     private static final char[] UNSUPPORTED_NAME_CHARACTERS = "!\" #$%&'()*+,./\\:;<=>?@[]^`{|}~".toCharArray();
+    private static final Pattern TOPIC_NAME_PATTERN = Pattern.compile("^[a-zA-Z0-9]([._-](?![._-])|"
+            + "[a-zA-Z0-9]){0,120}[a-zA-Z0-9]$");
 
     /**
      * Validate characters in names within cps.
@@ -48,4 +51,12 @@ public final class CpsValidator {
             }
         }
     }
+
+    /**
+     * Validate kafka topic name pattern.
+     * @param topicName name of the topic to be validated
+     */
+    public static boolean validateTopicName(final String topicName) {
+        return topicName != null && TOPIC_NAME_PATTERN.matcher(topicName).matches();
+    }
 }
diff --git a/cps-service/src/test/groovy/org/onap/cps/utils/CpsValidatorSpec.groovy b/cps-service/src/test/groovy/org/onap/cps/utils/CpsValidatorSpec.groovy
index 191472ceea..ce728ef1c1 100644
--- a/cps-service/src/test/groovy/org/onap/cps/utils/CpsValidatorSpec.groovy
+++ b/cps-service/src/test/groovy/org/onap/cps/utils/CpsValidatorSpec.groovy
@@ -45,4 +45,18 @@ class CpsValidatorSpec extends Specification {
             'position 5' | 'name with spaces' || 'name with spaces invalid token encountered at position 5'
             'position 9' | 'nameWith Space'   || 'nameWith Space invalid token encountered at position 9'
     }
+
+    def 'Validating topic names.'() {
+        when: 'the topic name is validated'
+            def isValidTopicName = CpsValidator.validateTopicName(topicName)
+        then: 'boolean response will be returned for #scenario'
+            assert isValidTopicName == booleanResponse
+        where: 'the following names are used'
+            scenario                  | topicName       || booleanResponse
+            'valid topic'             | 'my-topic-name' || true
+            'empty topic'             | ''              || false
+            'blank topic'             | ' '             || false
+            'null topic'              | null            || false
+            'invalid non empty topic' | '1_5_*_#'       || false
+    }
 }