Implement merging all ncmp datastore endpoints into one

- Merging all endpoints
under /v1/ch/{cm-handle}/data/ds/ncmp-datastore:*
to /v1/ch/{cm-handle}/data/ds/{ncmp-datastore-name}

- Implementing missing tests from parent
- Introducing abstract class to keep the common code and just pass in
  the supplier to be executed in sync or async manner
- Removed the existing get endpoints for passthrough-running,
  passthrough-operational and operational and merged them into a common
  get endpoint

Issue-ID: CPS-1178
Issue-ID: CPS-1001
Change-Id: I6956c81d5acfa8fb11217bcc16cb795b62070fa3
Signed-off-by: bmiklos <miklos.baranyak@est.tech>

10 files changed, 504 insertions, 109 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 d2ed393794..9aa8263fc5 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
@@ -29,21 +29,18 @@ import static org.onap.cps.ncmp.api.impl.operations.DmiRequestBody.OperationEnum
 import static org.onap.cps.ncmp.api.impl.operations.DmiRequestBody.OperationEnum.UPDATE;
 
 import java.util.List;
-import java.util.Map;
 import java.util.Set;
-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.inventory.CompositeState;
 import org.onap.cps.ncmp.api.models.CmHandleQueryApiParameters;
 import org.onap.cps.ncmp.api.models.NcmpServiceCmHandle;
 import org.onap.cps.ncmp.rest.api.NetworkCmProxyApi;
-import org.onap.cps.ncmp.rest.executor.CpsNcmpTaskExecutor;
+import org.onap.cps.ncmp.rest.controller.handlers.DatastoreType;
+import org.onap.cps.ncmp.rest.controller.handlers.NcmpDatastoreResourceRequestHandler;
+import org.onap.cps.ncmp.rest.controller.handlers.NcmpDatastoreResourceRequestHandlerFactory;
 import org.onap.cps.ncmp.rest.mapper.CmHandleStateMapper;
 import org.onap.cps.ncmp.rest.model.CmHandlePublicProperties;
 import org.onap.cps.ncmp.rest.model.CmHandleQueryParameters;
@@ -53,9 +50,7 @@ import org.onap.cps.ncmp.rest.model.RestOutputCmHandle;
 import org.onap.cps.ncmp.rest.model.RestOutputCmHandleCompositeState;
 import org.onap.cps.ncmp.rest.model.RestOutputCmHandlePublicProperties;
 import org.onap.cps.ncmp.rest.util.DeprecationHelper;
-import org.onap.cps.utils.CpsValidator;
 import org.onap.cps.utils.JsonObjectMapper;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -68,94 +63,50 @@ 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;
     private final NetworkCmProxyDataService networkCmProxyDataService;
     private final JsonObjectMapper jsonObjectMapper;
-
     private final DeprecationHelper deprecationHelper;
     private final NcmpRestInputMapper ncmpRestInputMapper;
     private final CmHandleStateMapper cmHandleStateMapper;
-    private final CpsNcmpTaskExecutor cpsNcmpTaskExecutor;
-    @Value("${notification.async.executor.time-out-value-in-ms:2000}")
-    private int timeOutInMilliSeconds;
-    @Value("${notification.enabled:true}")
-    private boolean asyncEnabled;
+    private final NcmpDatastoreResourceRequestHandlerFactory ncmpDatastoreResourceRequestHandlerFactory;
 
     /**
-     * Get resource data from operational datastore.
+     * Get resource data from datastore.
      *
-     * @param cmHandle cm handle identifier
-     * @param resourceIdentifier resource identifier
+     * @param datastoreName       name of the datastore
+     * @param cmHandle            cm handle identifier
+     * @param resourceIdentifier  resource identifier
      * @param optionsParamInQuery options query parameter
-     * @param topicParamInQuery topic query parameter
+     * @param topicParamInQuery   topic query parameter
+     * @param includeDescendants  whether include descendants
      * @return {@code ResponseEntity} response from dmi plugin
      */
-    @Override
-    public ResponseEntity<Object> getResourceDataOperationalForCmHandle(final String cmHandle,
-                                                                        final @NotNull @Valid String resourceIdentifier,
-                                                                        final @Valid String optionsParamInQuery,
-                                                                        final @Valid String topicParamInQuery) {
-        if (asyncEnabled && isValidTopic(topicParamInQuery)) {
-            final String requestId = UUID.randomUUID().toString();
-            log.info("Received Async passthrough-operational request with id {}", requestId);
-            cpsNcmpTaskExecutor.executeTask(() ->
-                    networkCmProxyDataService.getResourceDataOperationalForCmHandle(
-                        cmHandle, resourceIdentifier, optionsParamInQuery, topicParamInQuery, requestId
-                    ), timeOutInMilliSeconds
-            );
-            return ResponseEntity.ok(Map.of("requestId", requestId));
-        } else {
-            log.warn("Asynchronous messaging is currently disabled for passthrough-operational."
-                + " Will use synchronous operation.");
-        }
 
-        final Object responseObject = networkCmProxyDataService.getResourceDataOperationalForCmHandle(
-            cmHandle, resourceIdentifier, optionsParamInQuery, NO_TOPIC, NO_REQUEST_ID);
-
-        return ResponseEntity.ok(responseObject);
-    }
-
-    /**
-     * Get resource data from pass-through running datastore.
-     *
-     * @param cmHandle cm handle identifier
-     * @param resourceIdentifier resource identifier
-     * @param optionsParamInQuery options query parameter
-     * @param topicParamInQuery topic query parameter
-     * @return {@code ResponseEntity} response from dmi plugin
-     */
     @Override
-    public ResponseEntity<Object> getResourceDataRunningForCmHandle(final String cmHandle,
-                                                                    final @NotNull @Valid String resourceIdentifier,
-                                                                    final @Valid String optionsParamInQuery,
-                                                                    final @Valid String topicParamInQuery) {
-        if (asyncEnabled && isValidTopic(topicParamInQuery)) {
-            final String requestId = UUID.randomUUID().toString();
-            log.info("Received Async passthrough-running request with id {}", requestId);
-            cpsNcmpTaskExecutor.executeTask(() ->
-                networkCmProxyDataService.getResourceDataPassThroughRunningForCmHandle(
-                    cmHandle, resourceIdentifier, optionsParamInQuery, topicParamInQuery, requestId
-                ), timeOutInMilliSeconds
-            );
-            return ResponseEntity.ok(Map.of("requestId", requestId));
-        } else {
-            log.warn("Asynchronous messaging is currently disabled for passthrough-running."
-                + " Will use synchronous operation.");
-        }
+    public ResponseEntity<Object> getResourceDataForCmHandle(final String datastoreName,
+                                                             final String cmHandle,
+                                                             final String resourceIdentifier,
+                                                             final String optionsParamInQuery,
+                                                             final String topicParamInQuery,
+                                                             final Boolean includeDescendants) {
 
-        final Object responseObject = networkCmProxyDataService.getResourceDataPassThroughRunningForCmHandle(
-            cmHandle, resourceIdentifier, optionsParamInQuery, NO_TOPIC, NO_REQUEST_ID);
+        final NcmpDatastoreResourceRequestHandler ncmpDatastoreResourceRequestHandler =
+                ncmpDatastoreResourceRequestHandlerFactory.getNcmpDatastoreResourceRequestHandler(
+                        DatastoreType.fromDatastoreName(datastoreName));
 
-        return ResponseEntity.ok(responseObject);
+        return ncmpDatastoreResourceRequestHandler.getResourceData(cmHandle, resourceIdentifier,
+                optionsParamInQuery, topicParamInQuery, includeDescendants);
     }
 
     @Override
     public ResponseEntity<Object> patchResourceDataRunningForCmHandle(final String resourceIdentifier,
-        final String cmHandle,
-        final Object requestBody, final String contentType) {
-        final Object responseObject = networkCmProxyDataService.writeResourceDataPassThroughRunningForCmHandle(cmHandle,
-            resourceIdentifier, PATCH, jsonObjectMapper.asJsonString(requestBody), contentType);
+                                                                      final String cmHandle,
+                                                                      final Object requestBody,
+                                                                      final String contentType) {
+        final Object responseObject = networkCmProxyDataService
+                .writeResourceDataPassThroughRunningForCmHandle(
+                        cmHandle, resourceIdentifier, PATCH,
+                        jsonObjectMapper.asJsonString(requestBody), contentType);
         return ResponseEntity.ok(responseObject);
     }
 
@@ -163,14 +114,16 @@ public class NetworkCmProxyController implements NetworkCmProxyApi {
      * Create resource data in datastore pass-through running for given cm-handle.
      *
      * @param resourceIdentifier resource identifier
-     * @param cmHandle cm handle identifier
-     * @param requestBody the request body
-     * @param contentType content type of body
+     * @param cmHandle           cm handle identifier
+     * @param requestBody        the request body
+     * @param contentType        content type of body
      * @return {@code ResponseEntity} response from dmi plugin
      */
     @Override
     public ResponseEntity<Void> createResourceDataRunningForCmHandle(final String resourceIdentifier,
-        final String cmHandle, final Object requestBody, final String contentType) {
+                                                                     final String cmHandle,
+                                                                     final Object requestBody,
+                                                                     final String contentType) {
         networkCmProxyDataService.writeResourceDataPassThroughRunningForCmHandle(cmHandle,
                 resourceIdentifier, CREATE, jsonObjectMapper.asJsonString(requestBody), contentType);
         return new ResponseEntity<>(HttpStatus.CREATED);
@@ -180,9 +133,9 @@ public class NetworkCmProxyController implements NetworkCmProxyApi {
      * Update resource data in datastore pass-through running for given cm-handle.
      *
      * @param resourceIdentifier resource identifier
-     * @param cmHandle cm handle identifier
-     * @param requestBody the request body
-     * @param contentType content type of the body
+     * @param cmHandle           cm handle identifier
+     * @param requestBody        the request body
+     * @param contentType        content type of the body
      * @return response entity
      */
     @Override
@@ -197,11 +150,11 @@ public class NetworkCmProxyController implements NetworkCmProxyApi {
 
 
     /**
-     *  Delete resource data in datastore pass-through running for a given cm-handle.
+     * Delete resource data in datastore pass-through running for a given cm-handle.
      *
      * @param resourceIdentifier resource identifier
-     * @param cmHandle cm handle identifier
-     * @param contentType content type of the body
+     * @param cmHandle           cm handle identifier
+     * @param contentType        content type of the body
      * @return response entity no content if request is successful
      */
     @Override
@@ -209,7 +162,7 @@ public class NetworkCmProxyController implements NetworkCmProxyApi {
                                                                      final String resourceIdentifier,
                                                                      final String contentType) {
         networkCmProxyDataService.writeResourceDataPassThroughRunningForCmHandle(cmHandle,
-            resourceIdentifier, DELETE, NO_BODY, contentType);
+                resourceIdentifier, DELETE, NO_BODY, contentType);
         return new ResponseEntity<>(HttpStatus.NO_CONTENT);
     }
 
@@ -240,7 +193,7 @@ public class NetworkCmProxyController implements NetworkCmProxyApi {
      */
     @Override
     public ResponseEntity<List<String>> searchCmHandleIds(
-        final CmHandleQueryParameters cmHandleQueryParameters) {
+            final CmHandleQueryParameters cmHandleQueryParameters) {
         final CmHandleQueryApiParameters cmHandleQueryApiParameters =
                 jsonObjectMapper.convertToValueType(cmHandleQueryParameters, CmHandleQueryApiParameters.class);
         final Set<String> cmHandleIds = networkCmProxyDataService.executeCmHandleIdSearch(cmHandleQueryApiParameters);
@@ -249,6 +202,7 @@ public class NetworkCmProxyController implements NetworkCmProxyApi {
 
     /**
      * Search for Cm Handle and Properties by Name.
+     *
      * @param cmHandleId cm-handle identifier
      * @return cm handle and its properties
      */
@@ -261,33 +215,35 @@ public class NetworkCmProxyController implements NetworkCmProxyApi {
 
     /**
      * Get Cm Handle Properties by Cm Handle Id.
+     *
      * @param cmHandleId cm-handle identifier
      * @return cm handle properties
      */
     @Override
     public ResponseEntity<RestOutputCmHandlePublicProperties> getCmHandlePublicPropertiesByCmHandleId(
-        final String cmHandleId) {
+            final String cmHandleId) {
         final CmHandlePublicProperties cmHandlePublicProperties = new CmHandlePublicProperties();
         cmHandlePublicProperties.add(networkCmProxyDataService.getCmHandlePublicProperties(cmHandleId));
         final RestOutputCmHandlePublicProperties restOutputCmHandlePublicProperties =
-            new RestOutputCmHandlePublicProperties();
+                new RestOutputCmHandlePublicProperties();
         restOutputCmHandlePublicProperties.setPublicCmHandleProperties(cmHandlePublicProperties);
         return ResponseEntity.ok(restOutputCmHandlePublicProperties);
     }
 
     /**
      * Get Cm Handle State by Cm Handle Id.
+     *
      * @param cmHandleId cm-handle identifier
      * @return cm handle state
      */
     @Override
     public ResponseEntity<RestOutputCmHandleCompositeState> getCmHandleStateByCmHandleId(
-        final String cmHandleId) {
+            final String cmHandleId) {
         final CompositeState cmHandleState = networkCmProxyDataService.getCmHandleCompositeState(cmHandleId);
         final RestOutputCmHandleCompositeState restOutputCmHandleCompositeState =
-            new RestOutputCmHandleCompositeState();
+                new RestOutputCmHandleCompositeState();
         restOutputCmHandleCompositeState.setState(
-            cmHandleStateMapper.toCmHandleCompositeStateExternalLockReason(cmHandleState));
+                cmHandleStateMapper.toCmHandleCompositeStateExternalLockReason(cmHandleState));
         return ResponseEntity.ok(restOutputCmHandleCompositeState);
     }
 
@@ -314,22 +270,22 @@ public class NetworkCmProxyController implements NetworkCmProxyApi {
      */
     public ResponseEntity<List<RestModuleReference>> getModuleReferencesByCmHandle(final String cmHandle) {
         final List<RestModuleReference> restModuleReferences =
-            networkCmProxyDataService.getYangResourcesModuleReferences(cmHandle).stream()
-            .map(ncmpRestInputMapper::toRestModuleReference)
-                .collect(Collectors.toList());
+                networkCmProxyDataService.getYangResourcesModuleReferences(cmHandle).stream()
+                        .map(ncmpRestInputMapper::toRestModuleReference)
+                        .collect(Collectors.toList());
         return new ResponseEntity<>(restModuleReferences, HttpStatus.OK);
     }
 
     /**
      * Set the data sync enabled flag, along with the data sync state for the specified cm handle.
      *
-     * @param cmHandleId cm handle id
+     * @param cmHandleId          cm handle id
      * @param dataSyncEnabledFlag data sync enabled flag
      * @return response entity ok if request is successful
      */
     @Override
     public ResponseEntity<Object> setDataSyncEnabledFlagForCmHandle(final String cmHandleId,
-                                                                final Boolean dataSyncEnabledFlag) {
+                                                                    final Boolean dataSyncEnabledFlag) {
         networkCmProxyDataService.setDataSyncEnabled(cmHandleId, dataSyncEnabledFlag);
         return new ResponseEntity<>(HttpStatus.OK);
     }
@@ -345,15 +301,6 @@ public class NetworkCmProxyController implements NetworkCmProxyApi {
         return restOutputCmHandle;
     }
 
-    private static boolean isValidTopic(final String topicName) {
-        if (topicName == null) {
-            return false;
-        }
-        if (CpsValidator.validateTopicName(topicName)) {
-            return true;
-        }
-        throw new InvalidTopicException("Topic name " + topicName + " is invalid", "invalid topic");
-    }
 
 }
 
diff --git a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/DatastoreType.java b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/DatastoreType.java
new file mode 100644
index 0000000000..959c85d141
--- /dev/null
+++ b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/DatastoreType.java
@@ -0,0 +1,53 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2022 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.rest.controller.handlers;
+
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import lombok.Getter;
+
+@Getter
+public enum DatastoreType {
+
+    OPERATIONAL("ncmp-datastore:operational"),
+    PASSTHROUGH_RUNNING("ncmp-datastore:passthrough-running"),
+    PASSTHROUGH_OPERATIONAL("ncmp-datastore:passthrough-operational");
+
+    DatastoreType(final String datastoreName) {
+        this.datastoreName = datastoreName;
+    }
+
+    private final String datastoreName;
+    private static final Map<String, DatastoreType> datastoreNameToDatastoreType = new HashMap<>();
+
+    static {
+        Arrays.stream(DatastoreType.values()).forEach(
+                type -> datastoreNameToDatastoreType.put(type.getDatastoreName(), type));
+    }
+
+    public static DatastoreType fromDatastoreName(final String datastoreName) {
+        return datastoreNameToDatastoreType.get(datastoreName);
+    }
+
+}
+
diff --git a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastoreOperationalResourceRequestHandler.java b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastoreOperationalResourceRequestHandler.java
new file mode 100644
index 0000000000..6ed9b8c4d1
--- /dev/null
+++ b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastoreOperationalResourceRequestHandler.java
@@ -0,0 +1,55 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2022 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.rest.controller.handlers;
+
+import java.util.function.Supplier;
+import lombok.extern.slf4j.Slf4j;
+import org.onap.cps.ncmp.api.NetworkCmProxyDataService;
+import org.onap.cps.ncmp.rest.executor.CpsNcmpTaskExecutor;
+import org.onap.cps.spi.FetchDescendantsOption;
+
+@Slf4j
+public class NcmpDatastoreOperationalResourceRequestHandler extends NcmpDatastoreResourceRequestHandler {
+
+    public NcmpDatastoreOperationalResourceRequestHandler(final NetworkCmProxyDataService networkCmProxyDataService,
+                                                          final CpsNcmpTaskExecutor cpsNcmpTaskExecutor,
+                                                          final int timeOutInMilliSeconds,
+                                                          final boolean notificationFeatureEnabled) {
+        super(networkCmProxyDataService, cpsNcmpTaskExecutor, timeOutInMilliSeconds, notificationFeatureEnabled);
+    }
+
+    @Override
+    public Supplier<Object> getTask(final String cmHandle,
+                                    final String resourceIdentifier,
+                                    final String optionsParamInQuery,
+                                    final String topicParamInQuery,
+                                    final String requestId,
+                                    final Boolean includeDescendant) {
+
+        final FetchDescendantsOption fetchDescendantsOption =
+                Boolean.TRUE.equals(includeDescendant) ? FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS
+                        : FetchDescendantsOption.OMIT_DESCENDANTS;
+
+        return () -> networkCmProxyDataService.getResourceDataOperational(cmHandle, resourceIdentifier,
+                fetchDescendantsOption);
+    }
+
+}
diff --git a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastorePassthroughOperationalResourceRequestHandler.java b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastorePassthroughOperationalResourceRequestHandler.java
new file mode 100644
index 0000000000..196e5bd33d
--- /dev/null
+++ b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastorePassthroughOperationalResourceRequestHandler.java
@@ -0,0 +1,51 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2022 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.rest.controller.handlers;
+
+import java.util.function.Supplier;
+import lombok.extern.slf4j.Slf4j;
+import org.onap.cps.ncmp.api.NetworkCmProxyDataService;
+import org.onap.cps.ncmp.rest.executor.CpsNcmpTaskExecutor;
+
+@Slf4j
+public class NcmpDatastorePassthroughOperationalResourceRequestHandler extends NcmpDatastoreResourceRequestHandler {
+
+    public NcmpDatastorePassthroughOperationalResourceRequestHandler(
+            final NetworkCmProxyDataService networkCmProxyDataService,
+            final CpsNcmpTaskExecutor cpsNcmpTaskExecutor,
+            final int timeOutInMilliSeconds,
+            final boolean notificationFeatureEnabled) {
+        super(networkCmProxyDataService, cpsNcmpTaskExecutor, timeOutInMilliSeconds, notificationFeatureEnabled);
+    }
+
+    @Override
+    public Supplier<Object> getTask(final String cmHandle,
+                                    final String resourceIdentifier,
+                                    final String optionsParamInQuery,
+                                    final String topicParamInQuery,
+                                    final String requestId,
+                                    final Boolean includeDescendant) {
+
+        return () -> networkCmProxyDataService.getResourceDataOperationalForCmHandle(
+                cmHandle, resourceIdentifier, optionsParamInQuery, topicParamInQuery, requestId);
+    }
+
+}
diff --git a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastorePassthroughRunningResourceRequestHandler.java b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastorePassthroughRunningResourceRequestHandler.java
new file mode 100644
index 0000000000..5bf16b7499
--- /dev/null
+++ b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastorePassthroughRunningResourceRequestHandler.java
@@ -0,0 +1,50 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2022 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.rest.controller.handlers;
+
+import java.util.function.Supplier;
+import lombok.extern.slf4j.Slf4j;
+import org.onap.cps.ncmp.api.NetworkCmProxyDataService;
+import org.onap.cps.ncmp.rest.executor.CpsNcmpTaskExecutor;
+
+@Slf4j
+public class NcmpDatastorePassthroughRunningResourceRequestHandler extends NcmpDatastoreResourceRequestHandler {
+
+    public NcmpDatastorePassthroughRunningResourceRequestHandler(
+            final NetworkCmProxyDataService networkCmProxyDataService,
+            final CpsNcmpTaskExecutor cpsNcmpTaskExecutor,
+            final int timeOutInMilliSeconds,
+            final boolean notificationFeatureEnabled) {
+        super(networkCmProxyDataService, cpsNcmpTaskExecutor, timeOutInMilliSeconds, notificationFeatureEnabled);
+    }
+
+    @Override
+    public Supplier<Object> getTask(final String cmHandle,
+                                    final String resourceIdentifier,
+                                    final String optionsParamInQuery,
+                                    final String topicParamInQuery,
+                                    final String requestId,
+                                    final Boolean includeDescendant) {
+
+        return () -> networkCmProxyDataService.getResourceDataPassThroughRunningForCmHandle(
+                cmHandle, resourceIdentifier, optionsParamInQuery, topicParamInQuery, requestId);
+    }
+}
diff --git a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastoreResourceRequestHandler.java b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastoreResourceRequestHandler.java
new file mode 100644
index 0000000000..a6d313b05f
--- /dev/null
+++ b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastoreResourceRequestHandler.java
@@ -0,0 +1,92 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2022 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.rest.controller.handlers;
+
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Supplier;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.onap.cps.ncmp.api.NetworkCmProxyDataService;
+import org.onap.cps.ncmp.rest.executor.CpsNcmpTaskExecutor;
+import org.onap.cps.ncmp.rest.util.TopicValidator;
+import org.springframework.http.ResponseEntity;
+
+@RequiredArgsConstructor
+@Slf4j
+public abstract class NcmpDatastoreResourceRequestHandler {
+
+    private static final String NO_REQUEST_ID = null;
+    private static final String NO_TOPIC = null;
+
+    protected final NetworkCmProxyDataService networkCmProxyDataService;
+    protected final CpsNcmpTaskExecutor cpsNcmpTaskExecutor;
+    protected final int timeOutInMilliSeconds;
+    protected final boolean notificationFeatureEnabled;
+
+    protected abstract Supplier<Object> getTask(final String cmHandle,
+                                                final String resourceIdentifier,
+                                                final String optionsParamInQuery,
+                                                final String topicParamInQuery,
+                                                final String requestId,
+                                                final Boolean includeDescendant);
+
+
+    /**
+     * Get resource data from datastore.
+     *
+     * @param cmHandleId          the cm handle
+     * @param resourceIdentifier  the resource identifier
+     * @param optionsParamInQuery the options param in query
+     * @param topicParamInQuery   the topic param in query
+     * @param includeDescendants  whether include descendants
+     * @return the response entity
+     */
+    public ResponseEntity<Object> getResourceData(final String cmHandleId,
+                                                  final String resourceIdentifier,
+                                                  final String optionsParamInQuery,
+                                                  final String topicParamInQuery,
+                                                  final Boolean includeDescendants) {
+
+        final String requestId = UUID.randomUUID().toString();
+        final boolean asyncResponseRequested = topicParamInQuery != null;
+
+        if (asyncResponseRequested && notificationFeatureEnabled) {
+            TopicValidator.validateTopicName(topicParamInQuery);
+            log.debug("Received Async request with id {}", requestId);
+            cpsNcmpTaskExecutor.executeTask(
+                    getTask(cmHandleId, resourceIdentifier, optionsParamInQuery, topicParamInQuery, requestId,
+                            includeDescendants), timeOutInMilliSeconds);
+
+            return ResponseEntity.ok(Map.of("requestId", requestId));
+        }
+
+        if (asyncResponseRequested) {
+            log.warn("Asynchronous messaging is currently disabled, will use synchronous operation.");
+        }
+        
+        final Object responseObject =
+                getTask(cmHandleId, resourceIdentifier, optionsParamInQuery, NO_TOPIC, NO_REQUEST_ID,
+                        includeDescendants).get();
+
+        return ResponseEntity.ok(responseObject);
+    }
+}
diff --git a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastoreResourceRequestHandlerFactory.java b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastoreResourceRequestHandlerFactory.java
new file mode 100644
index 0000000000..35bd578ce2
--- /dev/null
+++ b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastoreResourceRequestHandlerFactory.java
@@ -0,0 +1,64 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2022 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.rest.controller.handlers;
+
+import lombok.RequiredArgsConstructor;
+import org.onap.cps.ncmp.api.NetworkCmProxyDataService;
+import org.onap.cps.ncmp.rest.executor.CpsNcmpTaskExecutor;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+@Component
+@RequiredArgsConstructor
+public class NcmpDatastoreResourceRequestHandlerFactory {
+
+    private final NetworkCmProxyDataService networkCmProxyDataService;
+    private final CpsNcmpTaskExecutor cpsNcmpTaskExecutor;
+
+    @Value("${notification.async.executor.time-out-value-in-ms:2000}")
+    private int timeOutInMilliSeconds;
+    @Value("${notification.enabled:true}")
+    private boolean notificationFeatureEnabled;
+
+    /**
+     * Gets ncmp datastore handler.
+     *
+     * @param datastoreType the datastore type
+     * @return the ncmp datastore handler
+     */
+    public NcmpDatastoreResourceRequestHandler getNcmpDatastoreResourceRequestHandler(
+            final DatastoreType datastoreType) {
+
+        switch (datastoreType) {
+            case OPERATIONAL:
+                return new NcmpDatastoreOperationalResourceRequestHandler(networkCmProxyDataService,
+                        cpsNcmpTaskExecutor, timeOutInMilliSeconds, notificationFeatureEnabled);
+            case PASSTHROUGH_RUNNING:
+                return new NcmpDatastorePassthroughRunningResourceRequestHandler(networkCmProxyDataService,
+                        cpsNcmpTaskExecutor, timeOutInMilliSeconds, notificationFeatureEnabled);
+            case PASSTHROUGH_OPERATIONAL:
+                return new NcmpDatastorePassthroughOperationalResourceRequestHandler(networkCmProxyDataService,
+                        cpsNcmpTaskExecutor, timeOutInMilliSeconds, notificationFeatureEnabled);
+            default:
+                return null;
+        }
+    }
+}
diff --git a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/exceptions/InvalidTopicException.java b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/exceptions/InvalidTopicException.java
new file mode 100644
index 0000000000..6a52d5861e
--- /dev/null
+++ b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/exceptions/InvalidTopicException.java
@@ -0,0 +1,40 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2022 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.rest.exceptions;
+
+import lombok.Getter;
+
+public class InvalidTopicException extends RuntimeException {
+
+    @Getter
+    final String details;
+
+    /**
+     * Constructor.
+     *
+     * @param message the error message
+     * @param details the error details
+     */
+    public InvalidTopicException(final String message, final String details) {
+        super(message);
+        this.details = details;
+    }
+}
diff --git a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/exceptions/NetworkCmProxyRestExceptionHandler.java b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/exceptions/NetworkCmProxyRestExceptionHandler.java
index c72373344d..98d7f6fd1d 100755
--- a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/exceptions/NetworkCmProxyRestExceptionHandler.java
+++ b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/exceptions/NetworkCmProxyRestExceptionHandler.java
@@ -25,7 +25,6 @@ import lombok.NoArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.onap.cps.ncmp.api.impl.exception.DmiRequestException;
 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.exception.NcmpException;
 import org.onap.cps.ncmp.api.impl.exception.ServerNcmpException;
 import org.onap.cps.ncmp.rest.controller.NetworkCmProxyController;
diff --git a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/util/TopicValidator.java b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/util/TopicValidator.java
new file mode 100644
index 0000000000..313e7bc012
--- /dev/null
+++ b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/util/TopicValidator.java
@@ -0,0 +1,44 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2022 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.rest.util;
+
+import java.util.regex.Pattern;
+import org.onap.cps.ncmp.rest.exceptions.InvalidTopicException;
+
+public class TopicValidator {
+
+    private static final Pattern TOPIC_NAME_PATTERN = Pattern.compile("^[a-zA-Z0-9]([._-](?![._-])|"
+        + "[a-zA-Z0-9]){0,120}[a-zA-Z0-9]$");
+
+    /**
+     * Validate kafka topic name pattern.
+     *
+     * @param topicName name of the topic to be validated
+     *
+     * @throws InvalidTopicException if the topic is not valid
+     */
+    public static void validateTopicName(final String topicName) {
+        if (!TOPIC_NAME_PATTERN.matcher(topicName).matches()) {
+            throw new InvalidTopicException("Topic name " + topicName + " is invalid", "invalid topic");
+        }
+    }
+
+}