Merge "Refactor ncmp request handlers (fix async issue)"

9 files changed, 371 insertions, 245 deletions

diff --git a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpCachedResourceRequestHandler.java b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpCachedResourceRequestHandler.java
index 76946d3af2..85a1eae234 100644
--- a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpCachedResourceRequestHandler.java
+++ b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpCachedResourceRequestHandler.java
@@ -25,6 +25,7 @@ import org.onap.cps.ncmp.api.NetworkCmProxyDataService;
 import org.onap.cps.ncmp.api.NetworkCmProxyQueryService;
 import org.onap.cps.ncmp.rest.executor.CpsNcmpTaskExecutor;
 import org.onap.cps.spi.FetchDescendantsOption;
+import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Component;
 
 @Component
@@ -48,37 +49,53 @@ public class NcmpCachedResourceRequestHandler extends NcmpDatastoreRequestHandle
         this.networkCmProxyQueryService = networkCmProxyQueryService;
     }
 
+    /**
+     * Executes a synchronous query request for given cm handle.
+     * Note. Currently only ncmp-datastore:operational supports query operations.
+     *
+     * @param cmHandleId         the cm handle
+     * @param resourceIdentifier the resource identifier
+     * @param includeDescendants whether include descendants
+     * @return the response entity
+     */
+    public ResponseEntity<Object> executeRequest(final String cmHandleId,
+                                                 final String resourceIdentifier,
+                                                 final boolean includeDescendants) {
+
+        final Supplier<Object> taskSupplier = getTaskSupplierForQueryRequest(cmHandleId, resourceIdentifier,
+            includeDescendants);
+        return executeTaskSync(taskSupplier);
+    }
+
     @Override
-    public Supplier<Object> getTaskSupplierForGetRequest(final String datastoreName,
-                                                         final String cmHandleId,
-                                                         final String resourceIdentifier,
-                                                         final String optionsParamInQuery,
-                                                         final String topicParamInQuery,
-                                                         final String requestId,
-                                                         final boolean includeDescendants) {
+    protected Supplier<Object> getTaskSupplierForGetRequest(final String datastoreName,
+                                                  final String cmHandleId,
+                                                  final String resourceIdentifier,
+                                                  final String optionsParamInQuery,
+                                                  final String topicParamInQuery,
+                                                  final String requestId,
+                                                  final boolean includeDescendants) {
 
-        final FetchDescendantsOption fetchDescendantsOption =
-                TaskManagementDefaultHandler.getFetchDescendantsOption(includeDescendants);
+        final FetchDescendantsOption fetchDescendantsOption = getFetchDescendantsOption(includeDescendants);
 
         return () -> networkCmProxyDataService.getResourceDataForCmHandle(datastoreName, cmHandleId, resourceIdentifier,
-                fetchDescendantsOption);
+            fetchDescendantsOption);
     }
 
-    /**
-     * Gets ncmp datastore query handler.
-     * Note. Currently only ncmp-datastore:operational supports query operations
-     * @return a ncmp datastore query handler.
-     */
-    @Override
-    public Supplier<Object> getTaskSupplierForQueryRequest(final String cmHandleId,
-                                                           final String resourceIdentifier,
-                                                           final boolean includeDescendants) {
+    private Supplier<Object> getTaskSupplierForQueryRequest(final String cmHandleId,
+                                                            final String resourceIdentifier,
+                                                            final boolean includeDescendants) {
 
-        final FetchDescendantsOption fetchDescendantsOption =
-                TaskManagementDefaultHandler.getFetchDescendantsOption(includeDescendants);
+        final FetchDescendantsOption fetchDescendantsOption = getFetchDescendantsOption(includeDescendants);
 
         return () -> networkCmProxyQueryService.queryResourceDataOperational(cmHandleId, resourceIdentifier,
-                fetchDescendantsOption);
+            fetchDescendantsOption);
     }
 
+    private static FetchDescendantsOption getFetchDescendantsOption(final boolean includeDescendants) {
+        return includeDescendants ? FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS
+            : FetchDescendantsOption.OMIT_DESCENDANTS;
+    }
+
+
 }
diff --git a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastoreRequestHandler.java b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastoreRequestHandler.java
index d7aeab6b0f..d40ab9b390 100644
--- a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastoreRequestHandler.java
+++ b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastoreRequestHandler.java
@@ -20,30 +20,24 @@
 
 package org.onap.cps.ncmp.rest.controller.handlers;
 
-import static org.onap.cps.ncmp.api.impl.operations.DatastoreType.OPERATIONAL;
-import static org.onap.cps.ncmp.api.impl.operations.OperationType.READ;
-
 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.impl.exception.InvalidDatastoreException;
-import org.onap.cps.ncmp.api.impl.operations.DatastoreType;
-import org.onap.cps.ncmp.api.impl.operations.OperationType;
-import org.onap.cps.ncmp.api.models.DataOperationRequest;
-import org.onap.cps.ncmp.rest.exceptions.OperationNotSupportedException;
 import org.onap.cps.ncmp.rest.executor.CpsNcmpTaskExecutor;
 import org.onap.cps.ncmp.rest.util.TopicValidator;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Service;
 
 @Slf4j
 @Service
 @RequiredArgsConstructor
-public class NcmpDatastoreRequestHandler implements TaskManagementDefaultHandler {
+public abstract class NcmpDatastoreRequestHandler {
+
+    private static final String NO_REQUEST_ID = null;
+    private static final String NO_TOPIC = null;
 
     @Value("${notification.async.executor.time-out-value-in-ms:2000}")
     protected int timeOutInMilliSeconds;
@@ -51,10 +45,10 @@ public class NcmpDatastoreRequestHandler implements TaskManagementDefaultHandler
     @Value("${notification.enabled:true}")
     protected boolean notificationFeatureEnabled;
 
-    private final CpsNcmpTaskExecutor cpsNcmpTaskExecutor;
+    protected final CpsNcmpTaskExecutor cpsNcmpTaskExecutor;
 
     /**
-     * Executes synchronous/asynchronous request for given cm handle.
+     * Executes synchronous/asynchronous get request for given cm handle.
      *
      * @param datastoreName       the name of the datastore
      * @param cmHandleId          the cm handle
@@ -86,46 +80,10 @@ public class NcmpDatastoreRequestHandler implements TaskManagementDefaultHandler
         return executeTaskSync(taskSupplier);
     }
 
-    /**
-     * Executes a synchronous request for given cm handle.
-     * Note. Currently only ncmp-datastore:operational supports query operations.
-     *
-     * @param cmHandleId         the cm handle
-     * @param resourceIdentifier the resource identifier
-     * @param includeDescendants whether include descendants
-     * @return the response entity
-     */
-    public ResponseEntity<Object> executeRequest(final String cmHandleId,
-                                                 final String resourceIdentifier,
-                                                 final boolean includeDescendants) {
-
-        final Supplier<Object> taskSupplier = getTaskSupplierForQueryRequest(cmHandleId, resourceIdentifier,
-                includeDescendants);
-        return executeTaskSync(taskSupplier);
-    }
 
-    /**
-     * Executes asynchronous request for group of cm handles to resource data.
-     *
-     * @param topicParamInQuery        the topic param in query
-     * @param dataOperationRequest     data operation request details for resource data
-     * @return the response entity
-     */
-    public ResponseEntity<Object> executeRequest(final String topicParamInQuery,
-                                                 final DataOperationRequest
-                                                         dataOperationRequest) {
-        validateDataOperationRequest(topicParamInQuery, dataOperationRequest);
-        if (!notificationFeatureEnabled) {
-            return ResponseEntity.ok(Map.of("status",
-                    "Asynchronous request is unavailable as notification feature is currently disabled."));
-        }
-        return getRequestIdAndSendDataOperationRequestToDmiService(topicParamInQuery, dataOperationRequest);
-    }
-
-    protected ResponseEntity<Object> executeTaskAsync(final String topicParamInQuery,
+    private ResponseEntity<Object> executeTaskAsync(final String topicParamInQuery,
                                                       final String requestId,
                                                       final Supplier<Object> taskSupplier) {
-
         TopicValidator.validateTopicName(topicParamInQuery);
         log.debug("Received Async request with id {}", requestId);
         cpsNcmpTaskExecutor.executeTask(taskSupplier, timeOutInMilliSeconds);
@@ -145,33 +103,15 @@ public class NcmpDatastoreRequestHandler implements TaskManagementDefaultHandler
         final String requestId = UUID.randomUUID().toString();
         final Supplier<Object> taskSupplier = getTaskSupplierForGetRequest(datastoreName, cmHandleId,
                 resourceIdentifier, optionsParamInQuery, topicParamInQuery, requestId, includeDescendants);
-        if (taskSupplier == NO_OBJECT_SUPPLIER) {
-            return new ResponseEntity<>(Map.of("status", "Unable to execute request as "
-                    + "datastore is not implemented."), HttpStatus.NOT_IMPLEMENTED);
-        }
         return executeTaskAsync(topicParamInQuery, requestId, taskSupplier);
     }
 
-    private ResponseEntity<Object> getRequestIdAndSendDataOperationRequestToDmiService(final String topicParamInQuery,
-                                                                                       final DataOperationRequest
-                                                                                       dataOperationRequest) {
-        final String requestId = UUID.randomUUID().toString();
-        sendDataOperationRequestAsynchronously(topicParamInQuery, dataOperationRequest, requestId);
-        return ResponseEntity.ok(Map.of("requestId", requestId));
-    }
+    protected abstract Supplier<Object> getTaskSupplierForGetRequest(final String datastoreName,
+                                                  final String cmHandleId,
+                                                  final String resourceIdentifier,
+                                                  final String optionsParamInQuery,
+                                                  final String topicParamInQuery,
+                                                  final String requestId,
+                                                  final boolean includeDescendant);
 
-    private void validateDataOperationRequest(final String topicParamInQuery,
-                                              final DataOperationRequest
-                                              dataOperationRequest) {
-        TopicValidator.validateTopicName(topicParamInQuery);
-        dataOperationRequest.getDataOperationDefinitions().forEach(dataOperationDetail -> {
-            if (OperationType.fromOperationName(dataOperationDetail.getOperation()) != READ) {
-                throw new OperationNotSupportedException(
-                        dataOperationDetail.getOperation() + " operation not yet supported");
-            } else if (DatastoreType.fromDatastoreName(dataOperationDetail.getDatastore()) == OPERATIONAL) {
-                throw new InvalidDatastoreException(dataOperationDetail.getDatastore()
-                        + " datastore is not supported");
-            }
-        });
-    }
 }
diff --git a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpPassthroughResourceRequestHandler.java b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpPassthroughResourceRequestHandler.java
index 0e49c6df13..8a3257576d 100644
--- a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpPassthroughResourceRequestHandler.java
+++ b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/NcmpPassthroughResourceRequestHandler.java
@@ -20,11 +20,21 @@
 
 package org.onap.cps.ncmp.rest.controller.handlers;
 
+import static org.onap.cps.ncmp.api.impl.operations.DatastoreType.OPERATIONAL;
+import static org.onap.cps.ncmp.api.impl.operations.OperationType.READ;
+
+import java.util.Map;
+import java.util.UUID;
 import java.util.function.Supplier;
 import org.onap.cps.ncmp.api.NetworkCmProxyDataService;
+import org.onap.cps.ncmp.api.impl.exception.InvalidDatastoreException;
+import org.onap.cps.ncmp.api.impl.operations.DatastoreType;
+import org.onap.cps.ncmp.api.impl.operations.OperationType;
 import org.onap.cps.ncmp.api.models.DataOperationRequest;
+import org.onap.cps.ncmp.rest.exceptions.OperationNotSupportedException;
 import org.onap.cps.ncmp.rest.executor.CpsNcmpTaskExecutor;
-import org.springframework.scheduling.annotation.Async;
+import org.onap.cps.ncmp.rest.util.TopicValidator;
+import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Component;
 
 @Component
@@ -32,6 +42,8 @@ public class NcmpPassthroughResourceRequestHandler extends NcmpDatastoreRequestH
 
     private final NetworkCmProxyDataService networkCmProxyDataService;
 
+    private static final Object noReturn = null;
+
     /**
      * Constructor.
      *
@@ -44,27 +56,71 @@ public class NcmpPassthroughResourceRequestHandler extends NcmpDatastoreRequestH
         this.networkCmProxyDataService = networkCmProxyDataService;
     }
 
+    /**
+     * Executes asynchronous request for group of cm handles to resource data.
+     *
+     * @param topicParamInQuery        the topic param in query
+     * @param dataOperationRequest     data operation request details for resource data
+     * @return the response entity
+     */
+    public ResponseEntity<Object> executeRequest(final String topicParamInQuery,
+                                                 final DataOperationRequest
+                                                     dataOperationRequest) {
+        validateDataOperationRequest(topicParamInQuery, dataOperationRequest);
+        if (!notificationFeatureEnabled) {
+            return ResponseEntity.ok(Map.of("status",
+                "Asynchronous request is unavailable as notification feature is currently disabled."));
+        }
+        return getRequestIdAndSendDataOperationRequestToDmiService(topicParamInQuery, dataOperationRequest);
+    }
+
     @Override
-    public Supplier<Object> getTaskSupplierForGetRequest(final String datastoreName,
-                                                         final String cmHandleId,
-                                                         final String resourceIdentifier,
-                                                         final String optionsParamInQuery,
-                                                         final String topicParamInQuery,
-                                                         final String requestId,
-                                                         final boolean includeDescendants) {
+    protected Supplier<Object> getTaskSupplierForGetRequest(final String datastoreName,
+                                                            final String cmHandleId,
+                                                            final String resourceIdentifier,
+                                                            final String optionsParamInQuery,
+                                                            final String topicParamInQuery,
+                                                            final String requestId,
+                                                            final boolean includeDescendants) {
 
         return () -> networkCmProxyDataService.getResourceDataForCmHandle(
-                datastoreName, cmHandleId, resourceIdentifier, optionsParamInQuery, topicParamInQuery, requestId);
+            datastoreName, cmHandleId, resourceIdentifier, optionsParamInQuery, topicParamInQuery, requestId);
     }
 
-    @Async
-    @Override
-    public void sendDataOperationRequestAsynchronously(final String topicParamInQuery,
-                                                       final DataOperationRequest
-                                                                   dataOperationRequest,
-                                                       final String requestId) {
-        networkCmProxyDataService.executeDataOperationForCmHandles(topicParamInQuery, dataOperationRequest,
-                requestId);
+    private ResponseEntity<Object> getRequestIdAndSendDataOperationRequestToDmiService(final String topicParamInQuery,
+                                                                                       final DataOperationRequest
+                                                                                           dataOperationRequest) {
+        final String requestId = UUID.randomUUID().toString();
+        cpsNcmpTaskExecutor.executeTask(
+            getTaskSupplierForDataOperationRequest(topicParamInQuery, dataOperationRequest, requestId),
+            timeOutInMilliSeconds);
+        return ResponseEntity.ok(Map.of("requestId", requestId));
+    }
 
+    private void validateDataOperationRequest(final String topicParamInQuery,
+                                              final DataOperationRequest
+                                                  dataOperationRequest) {
+        TopicValidator.validateTopicName(topicParamInQuery);
+        dataOperationRequest.getDataOperationDefinitions().forEach(dataOperationDetail -> {
+            if (OperationType.fromOperationName(dataOperationDetail.getOperation()) != READ) {
+                throw new OperationNotSupportedException(
+                    dataOperationDetail.getOperation() + " operation not yet supported");
+            } else if (DatastoreType.fromDatastoreName(dataOperationDetail.getDatastore()) == OPERATIONAL) {
+                throw new InvalidDatastoreException(dataOperationDetail.getDatastore()
+                    + " datastore is not supported");
+            }
+        });
     }
+
+    private Supplier<Object> getTaskSupplierForDataOperationRequest(final String topicParamInQuery,
+                                                                    final DataOperationRequest dataOperationRequest,
+                                                                    final String requestId) {
+        return () -> {
+            networkCmProxyDataService.executeDataOperationForCmHandles(topicParamInQuery,
+                dataOperationRequest,
+                requestId);
+            return noReturn;
+        };
+    }
+
 }
diff --git a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/TaskManagementDefaultHandler.java b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/TaskManagementDefaultHandler.java
deleted file mode 100644
index b2520b1609..0000000000
--- a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/handlers/TaskManagementDefaultHandler.java
+++ /dev/null
@@ -1,59 +0,0 @@
-/*
- *  ============LICENSE_START=======================================================
- *  Copyright (C) 2023 Nordix Foundation
- *  ================================================================================
- *  Licensed under the Apache License, Version 2.0 (the "License");
- *  you may not use this file except in compliance with the License.
- *  You may obtain a copy of the License at
- *
- *        http://www.apache.org/licenses/LICENSE-2.0
- *
- *  Unless required by applicable law or agreed to in writing, software
- *  distributed under the License is distributed on an "AS IS" BASIS,
- *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
- *
- *  SPDX-License-Identifier: Apache-2.0
- *  ============LICENSE_END=========================================================
- */
-
-package org.onap.cps.ncmp.rest.controller.handlers;
-
-import java.util.function.Supplier;
-import org.onap.cps.ncmp.api.models.DataOperationRequest;
-import org.onap.cps.spi.FetchDescendantsOption;
-
-public interface TaskManagementDefaultHandler {
-
-    String NO_REQUEST_ID = null;
-    String NO_TOPIC = null;
-    Supplier<Object> NO_OBJECT_SUPPLIER = null;
-
-    default Supplier<Object> getTaskSupplierForGetRequest(final String datastoreName,
-                                                          final String cmHandleId,
-                                                          final String resourceIdentifier,
-                                                          final String optionsParamInQuery,
-                                                          final String topicParamInQuery,
-                                                          final String requestId,
-                                                          final boolean includeDescendant) {
-        return NO_OBJECT_SUPPLIER;
-    }
-
-    default Supplier<Object> getTaskSupplierForQueryRequest(final String cmHandleId,
-                                                            final String resourceIdentifier,
-                                                            final boolean includeDescendant) {
-        return NO_OBJECT_SUPPLIER;
-    }
-
-    default void sendDataOperationRequestAsynchronously(final String topicParamInQuery,
-                                                        final DataOperationRequest
-                                                                    dataOperationRequest,
-                                                        final String requestId) {
-    }
-
-    static FetchDescendantsOption getFetchDescendantsOption(final boolean includeDescendants) {
-        return includeDescendants ? FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS
-                : FetchDescendantsOption.OMIT_DESCENDANTS;
-    }
-}
diff --git a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/executor/CpsNcmpTaskExecutor.java b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/executor/CpsNcmpTaskExecutor.java
index 0543c4fba3..ba68d5b757 100644
--- a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/executor/CpsNcmpTaskExecutor.java
+++ b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/executor/CpsNcmpTaskExecutor.java
@@ -1,6 +1,6 @@
 /*
  *  ============LICENSE_START=======================================================
- *  Copyright (C) 2022 Nordix Foundation
+ *  Copyright (C) 2022-2023 Nordix Foundation
  *  ================================================================================
  *  Licensed under the Apache License, Version 2.0 (the "License");
  *  you may not use this file except in compliance with the License.
@@ -35,19 +35,19 @@ public class CpsNcmpTaskExecutor {
      * Execute a task asynchronously.
      *
      * @param taskSupplier functional method is get() task need to executed asynchronously
-     * @param timeOutInMillis the time out value in milliseconds
+     * @param timeOutInMillis the time-out value in milliseconds
      */
-    public void executeTask(final Supplier<Object> taskSupplier, final int timeOutInMillis) {
+    public void executeTask(final Supplier<Object> taskSupplier, final long timeOutInMillis) {
         CompletableFuture.supplyAsync(taskSupplier::get)
             .orTimeout(timeOutInMillis, MILLISECONDS)
-            .whenCompleteAsync((responseAsJson, throwable) -> handleTaskCompletion(throwable));
+            .whenCompleteAsync((taskResult, throwable) -> handleTaskCompletion(throwable));
     }
 
     private void handleTaskCompletion(final Throwable throwable) {
         if (throwable == null) {
             log.info("Async task completed successfully.");
         } else {
-            log.error("Async task failed. caused by : {}", throwable.getMessage());
+            log.error("Async task failed. caused by : {}", throwable.toString());
         }
     }
 }
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 4ee31e1ec5..7964e32b6f 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
@@ -75,7 +75,6 @@ import static org.onap.cps.ncmp.api.impl.operations.DatastoreType.OPERATIONAL
 import static org.onap.cps.spi.FetchDescendantsOption.OMIT_DESCENDANTS;
 import static org.onap.cps.spi.FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS;
 
-
 @WebMvcTest(NetworkCmProxyController)
 class NetworkCmProxyControllerSpec extends Specification {
 
@@ -104,16 +103,16 @@ class NetworkCmProxyControllerSpec extends Specification {
     DataOperationRequestMapper dataOperationRequestMapper = Mappers.getMapper(DataOperationRequestMapper)
 
     @SpringBean
-    CpsNcmpTaskExecutor spiedCpsTaskExecutor = Spy()
+    CpsNcmpTaskExecutor mockCpsTaskExecutor = Mock()
 
     @SpringBean
     DeprecationHelper stubbedDeprecationHelper = Stub()
 
     @SpringBean
-    NcmpCachedResourceRequestHandler ncmpCachedResourceRequestHandler = new NcmpCachedResourceRequestHandler(spiedCpsTaskExecutor, mockNetworkCmProxyDataService, mockNetworkCmProxyQueryService)
+    NcmpCachedResourceRequestHandler ncmpCachedResourceRequestHandler = new NcmpCachedResourceRequestHandler(mockCpsTaskExecutor, mockNetworkCmProxyDataService, mockNetworkCmProxyQueryService)
 
     @SpringBean
-    NcmpPassthroughResourceRequestHandler ncmpPassthroughResourceRequestHandler = new NcmpPassthroughResourceRequestHandler(spiedCpsTaskExecutor, mockNetworkCmProxyDataService)
+    NcmpPassthroughResourceRequestHandler ncmpPassthroughResourceRequestHandler = new NcmpPassthroughResourceRequestHandler(mockCpsTaskExecutor, mockNetworkCmProxyDataService)
 
     @Value('${rest.api.ncmp-base-path}/v1')
     def ncmpBasePathV1
@@ -150,23 +149,6 @@ class NetworkCmProxyControllerSpec extends Specification {
             response.status == HttpStatus.OK.value()
     }
 
-    def 'Get Resource Data Async Topic Handling with #scenario.'() {
-        given: 'resource data url'
-            def getUrl = "$ncmpBasePathV1/ch/testCmHandle/data/ds/ncmp-datastore:passthrough-operational?resourceIdentifier=parent/child&${topicQueryParam}"
-        when: 'get data resource request is performed'
-            def response = mvc.perform(
-                get(getUrl).contentType(MediaType.APPLICATION_JSON)).andReturn().response
-        then: 'task executor is called appropriate number of times'
-            expectedNumberOfTaskExecutions * spiedCpsTaskExecutor.executeTask(_, TIMOUT_FOR_TEST)
-        and: 'response status is OK'
-            response.status == HttpStatus.OK.value()
-        where: 'the following parameters are used'
-            scenario               | datastoreInUrl            | topicQueryParam        || expectedNumberOfTaskExecutions
-            'url with valid topic' | 'passthrough-operational' | '&topic=my-topic-name' || 1
-            'no topic in url'      | 'passthrough-operational' | ''                     || 0
-            'null topic in url'    | 'passthrough-operational' | '&topic=null'          || 1
-    }
-
     def 'Get Resource Data from ncmp-datastore:operational (cached) parameters handling with #scenario.'() {
         given: 'resource data url'
             def getUrl = "$ncmpBasePathV1/ch/h123/data/ds/ncmp-datastore:operational" +
@@ -188,30 +170,10 @@ class NetworkCmProxyControllerSpec extends Specification {
             'options (ignored)'         | '&options=(a-=1)'            || OMIT_DESCENDANTS
     }
 
-    def 'Get Resource Data with invalid topic parameter: #scenario.'() {
-        given: 'resource data url'
-            def getUrl = "$ncmpBasePathV1/ch/testCmHandle/data/ds/ncmp-datastore:${datastoreInUrl}" +
-                "?resourceIdentifier=parent/child&options=(a=1,b=2)${topicQueryParam}"
-        when: 'get data resource (async) request is performed'
-            def response = mvc.perform(
-                get(getUrl).contentType(MediaType.APPLICATION_JSON)).andReturn().response
-        then: 'abad request is returned'
-            response.status == HttpStatus.BAD_REQUEST.value()
-        where: 'the following parameters are used'
-            scenario                               | datastoreInUrl            | topicQueryParam
-            'empty topic in url'                   | 'passthrough-operational' | '&topic=\"\"'
-            'missing topic in url'                 | 'passthrough-operational' | '&topic='
-            'blank topic value in url'             | 'passthrough-operational' | '&topic=\" \"'
-            'invalid non-empty topic value in url' | 'passthrough-operational' | '&topic=1_5_*_#'
-    }
-
     def 'Execute (async) data operation to read data from dmi service.'() {
         given: 'data operation url'
             def getUrl = "$ncmpBasePathV1/data?topic=my-topic-name"
-            def dataOperationRequestJsonData = jsonObjectMapper.asJsonString(
-                    getDataOperationRequest("read", datastore.datastoreName))
-            def expectedDmiDataOperationRequest
-                    = jsonObjectMapper.convertJsonString(dataOperationRequestJsonData, org.onap.cps.ncmp.api.models.DataOperationRequest.class)
+            def dataOperationRequestJsonData = jsonObjectMapper.asJsonString(getDataOperationRequest("read", datastore.datastoreName))
         when: 'post data operation request is performed'
             def response = mvc.perform(
                     post(getUrl)
@@ -221,20 +183,18 @@ class NetworkCmProxyControllerSpec extends Specification {
         then: 'response status is Ok'
             response.status == HttpStatus.OK.value()
         and: 'async request id is generated'
-            assert response.contentAsString.contains("requestId")
-        then: 'wait a little to allow execution of service method by task executor (on separate thread)'
-            Thread.sleep(100)
-        then: 'the service has been invoked with the correct parameters '
-            1 * mockNetworkCmProxyDataService.executeDataOperationForCmHandles('my-topic-name', expectedDmiDataOperationRequest, _)
+            assert response.contentAsString.contains('requestId')
+        then: 'the request is handled asynchronously'
+            1 * mockCpsTaskExecutor.executeTask(*_)
         where: 'the following data stores are used'
             datastore << [PASSTHROUGH_RUNNING, PASSTHROUGH_OPERATIONAL]
     }
 
-    def 'Execute (async) data operation for #scenario from dmi service.'() {
+    def 'Execute (async) data operation with some validation error.'() {
         given: 'data operation url'
             def getUrl = "$ncmpBasePathV1/data?topic=my-topic-name"
             def dataOperationRequestJsonData = jsonObjectMapper.asJsonString(
-                    getDataOperationRequest(operation, datastore))
+                    getDataOperationRequest('read', 'invalid datastore'))
         when: 'post data resource request is performed'
             def response = mvc.perform(
                     post(getUrl)
@@ -243,18 +203,13 @@ class NetworkCmProxyControllerSpec extends Specification {
             ).andReturn().response
         then: 'response status is BAD_REQUEST'
             response.status == HttpStatus.BAD_REQUEST.value()
-        where: 'the following parameters are used'
-            scenario                                            | datastore                             | operation
-            'non-supported datastoreName'                       | OPERATIONAL.datastoreName             | 'read'
-            'non-supported operation (passthrough-running)'     | PASSTHROUGH_RUNNING.datastoreName     | 'create'
-            'non-supported operation (passthrough-operational)' | PASSTHROUGH_OPERATIONAL.datastoreName | 'create'
     }
 
     def 'Get data operation resource data when notification feature is disabled for datastore: #datastore.'() {
         given: 'data operation url'
             def getUrl = "$ncmpBasePathV1/data?topic=my-topic-name"
             def dataOperationRequestJsonData = jsonObjectMapper.asJsonString(
-                    getDataOperationRequest("read", datastore.datastoreName))
+                    getDataOperationRequest("read", PASSTHROUGH_RUNNING.datastoreName))
             ncmpPassthroughResourceRequestHandler.notificationFeatureEnabled = false
         when: 'post data resource request is performed'
             def response = mvc.perform(
@@ -266,8 +221,6 @@ class NetworkCmProxyControllerSpec extends Specification {
             response.status == HttpStatus.OK.value()
         and: 'async request id is unavailable'
             assert response.contentAsString == '{"status":"Asynchronous request is unavailable as notification feature is currently disabled."}'
-        where: 'the following data stores are used'
-            datastore << [PASSTHROUGH_RUNNING, PASSTHROUGH_OPERATIONAL]
     }
 
     def 'Query Resource Data from operational.'() {
@@ -287,9 +240,9 @@ class NetworkCmProxyControllerSpec extends Specification {
             response.status == HttpStatus.OK.value()
     }
 
-    def 'Query Resource Data using datastore of #datastore'() {
+    def 'Query Resource Data with unsupported datastore'() {
         given: 'the query resource data url'
-            def getUrl = "$ncmpBasePathV1/ch/testCmHandle/data/ds/ncmp-datastore:${datastore}/query" +
+            def getUrl = "$ncmpBasePathV1/ch/testCmHandle/data/ds/ncmp-datastore:passthrough-running/query" +
                 "?cps-path=/cps/path"
         when: 'the query data resource request is performed'
             def response = mvc.perform(
@@ -298,10 +251,8 @@ class NetworkCmProxyControllerSpec extends Specification {
             ).andReturn().response
         then: 'a 400 BAD_REQUEST is returned for the unsupported datastore'
             response.status == 400
-        and: 'the error message is that datastore #datastore is not supported'
-            response.contentAsString.contains("ncmp-datastore:${datastore} is not supported")
-        where: 'the following datastore is used'
-            datastore << ["passthrough-running", "passthrough-operational"]
+        and: 'the error message is that the datastore is not supported'
+            response.contentAsString.contains("ncmp-datastore:passthrough-running is not supported")
     }
 
     def 'Get Resource Data from pass-through running with #scenario value in resource identifier param.'() {
@@ -586,7 +537,7 @@ class NetworkCmProxyControllerSpec extends Specification {
     def 'Get Resource Data from operational with or without descendants'() {
         given: 'resource data url with descendants #enabled'
             def getUrl = "$ncmpBasePathV1/ch/testCmHandle/data/ds/ncmp-datastore:operational" +
-                "?resourceIdentifier=parent/child&include-descendants=${enabled}"
+                "?resourceIdentifier=parent/child&include-descendants=${booleanValue}"
         when: 'get data resource request is performed'
             def response = mvc.perform(
                 get(getUrl)
@@ -597,9 +548,9 @@ class NetworkCmProxyControllerSpec extends Specification {
         and: 'response status is Ok'
             response.status == HttpStatus.OK.value()
         where: 'the following parameters are used'
-            enabled | descendantsOption
-            false   | FetchDescendantsOption.OMIT_DESCENDANTS
-            true    | FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS
+            booleanValue | descendantsOption
+            false        | OMIT_DESCENDANTS
+            true         | INCLUDE_ALL_DESCENDANTS
     }
 
     def 'Attempt execute #operation rest operation on resource data with #scenario'() {
diff --git a/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/controller/NetworkCmProxyInventoryControllerSpec.groovy b/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/controller/NetworkCmProxyInventoryControllerSpec.groovy
index 300026d34a..e755094dd2 100644
--- a/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/controller/NetworkCmProxyInventoryControllerSpec.groovy
+++ b/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/controller/NetworkCmProxyInventoryControllerSpec.groovy
@@ -251,7 +251,7 @@ class NetworkCmProxyInventoryControllerSpec extends Specification {
     }
 
     def failedResponse(cmHandle) {
-        return CmHandleRegistrationResponse.createFailureResponse(cmHandle, new RuntimeException("Failed"))
+        return CmHandleRegistrationResponse.createFailureResponse(cmHandle, new RuntimeException('Failed'))
     }
 
     def successResponse(cmHandle) {
diff --git a/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastoreRequestHandlerSpec.groovy b/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastoreRequestHandlerSpec.groovy
new file mode 100644
index 0000000000..e149cb7c4a
--- /dev/null
+++ b/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/controller/handlers/NcmpDatastoreRequestHandlerSpec.groovy
@@ -0,0 +1,135 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2023 Nordix Foundation
+ *  ================================================================================
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *  ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.rest.controller.handlers
+
+import org.onap.cps.ncmp.api.NetworkCmProxyDataService
+import org.onap.cps.ncmp.api.impl.exception.InvalidDatastoreException
+import org.onap.cps.ncmp.api.impl.exception.InvalidOperationException
+import org.onap.cps.ncmp.api.models.DataOperationDefinition
+import org.onap.cps.ncmp.api.models.DataOperationRequest
+import org.onap.cps.ncmp.rest.exceptions.OperationNotSupportedException
+import org.onap.cps.ncmp.rest.executor.CpsNcmpTaskExecutor
+import spock.lang.Specification
+import spock.util.concurrent.PollingConditions
+
+class NcmpDatastoreRequestHandlerSpec extends Specification {
+
+    def spiedCpsNcmpTaskExecutor = Spy(CpsNcmpTaskExecutor)
+    def mockNetworkCmProxyDataService = Mock(NetworkCmProxyDataService)
+
+    def objectUnderTest = new NcmpPassthroughResourceRequestHandler(spiedCpsNcmpTaskExecutor, mockNetworkCmProxyDataService)
+
+    def 'Attempt to execute async get request with #scenario.'() {
+        given: 'notification feature is turned on/off'
+            objectUnderTest.notificationFeatureEnabled = notificationFeatureEnabled
+        when: 'get request is executed with topic = #topic'
+            objectUnderTest.executeRequest('ds', 'ch1', 'resource1', 'options', topic, false)
+        and: 'wait a little for async execution (only if expected)'
+            if (expectedCalls > 0) {
+                Thread.sleep(100)
+            }
+        then: 'the task is executed in an async fashion or not'
+            expectedCalls * spiedCpsNcmpTaskExecutor.executeTask(*_)
+        and: 'the service request is always invoked'
+            1 * mockNetworkCmProxyDataService.getResourceDataForCmHandle('ds', 'ch1', 'resource1', 'options', _, _)
+        where: 'the following parameters are used'
+            scenario                   | notificationFeatureEnabled | topic   || expectedCalls
+            'feature on, valid topic'  | true                       | 'valid' || 1
+            'feature on, no topic'     | true                       | null    || 0
+            'feature off, valid topic' | false                      | 'valid' || 0
+            'feature off, no topic'    | false                      | null    || 0
+    }
+
+    def 'Attempt to execute async data operation request with feature #scenario.'() {
+        given: 'a extended request handler that supports bulk requests'
+           def objectUnderTest = new NcmpPassthroughResourceRequestHandler(spiedCpsNcmpTaskExecutor, mockNetworkCmProxyDataService)
+        and: 'notification feature is turned on/off'
+            objectUnderTest.notificationFeatureEnabled = notificationFeatureEnabled
+        when: 'data operation request is executed'
+            objectUnderTest.executeRequest('someTopic', new DataOperationRequest())
+        then: 'the task is executed in an async fashion or not'
+            expectedCalls * spiedCpsNcmpTaskExecutor.executeTask(*_)
+        where: 'the following parameters are used'
+            scenario | notificationFeatureEnabled || expectedCalls
+            'on'     | true                       || 1
+            'off'    | false                      || 0
+    }
+
+    def 'Execute async data operation request with datastore #datastore.'() {
+        given: 'notification feature is turned on'
+            objectUnderTest.notificationFeatureEnabled = true
+        and: 'a data operation request with datastore: #datastore'
+            def dataOperationDefinition = new DataOperationDefinition(operation: 'read', datastore: datastore)
+            def dataOperationRequest = new DataOperationRequest(dataOperationDefinitions: [dataOperationDefinition])
+        and: ' a flag to track the network service call'
+            def networkServiceMethodCalled = false
+        and: 'the (mocked) service will use the flag to indicate it is called'
+            mockNetworkCmProxyDataService.executeDataOperationForCmHandles('myTopic', dataOperationRequest, _) >> {
+                networkServiceMethodCalled = true
+            }
+        when: 'data operation request is executed'
+            objectUnderTest.executeRequest('myTopic', dataOperationRequest)
+        then: 'the task is executed in an async fashion'
+            1 * spiedCpsNcmpTaskExecutor.executeTask(*_)
+        and: 'the network service is invoked (wait max. 5 seconds)'
+            new PollingConditions(timeout: 5).eventually {
+                networkServiceMethodCalled == true
+            }
+        where: 'the following datastores are used'
+            datastore << ['ncmp-datastore:passthrough-running', 'ncmp-datastore:passthrough-operational']
+    }
+
+    def 'Attempt to execute async data operation request with error #scenario'() {
+        given: 'notification feature is turned on'
+            objectUnderTest.notificationFeatureEnabled = true
+        and: 'a data operation definition with datastore: #datastore'
+            def dataOperationDefinition = new DataOperationDefinition(operation: 'read', datastore: datastore)
+        when: 'data operation request is executed'
+            def dataOperationRequest = new DataOperationRequest(dataOperationDefinitions: [dataOperationDefinition])
+            objectUnderTest.executeRequest('myTopic', dataOperationRequest)
+        then: 'the correct error is thrown'
+            def thrown = thrown(InvalidDatastoreException)
+            assert thrown.message.contains(expectedErrorMessage)
+        where: 'the following datastore names are used'
+            scenario                | datastore                    || expectedErrorMessage
+            'unsupported datastore' | 'ncmp-datastore:operational' || 'not supported'
+            'invalid datastore'     | 'invalid'                    || 'invalid datastore name'
+    }
+
+    def 'Attempt to execute async data operation request with #scenario operation: #operation.'() {
+        given: 'notification feature is turned on'
+            objectUnderTest.notificationFeatureEnabled = true
+        and: 'a data operation definition with operation: #operation'
+            def dataOperationDefinition = new DataOperationDefinition(operation: operation, datastore: 'ncmp-datastore:passthrough-running')
+        when: 'bulk request is executed'
+            objectUnderTest.executeRequest('someTopic', new DataOperationRequest(dataOperationDefinitions:[dataOperationDefinition]))
+        then: 'the expected type of exception is thrown'
+            thrown(expectedException)
+        where: 'the following operations are used'
+            scenario      | operation || expectedException
+            'invalid'     | 'invalid' || InvalidOperationException
+            'unsupported' | 'create'  || OperationNotSupportedException
+            'unsupported' | 'update'  || OperationNotSupportedException
+            'unsupported' | 'patch'   || OperationNotSupportedException
+            'unsupported' | 'delete'  || OperationNotSupportedException
+    }
+
+}
diff --git a/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/executor/CpsNcmpTaskExecutorSpec.groovy b/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/executor/CpsNcmpTaskExecutorSpec.groovy
new file mode 100644
index 0000000000..870c36c777
--- /dev/null
+++ b/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/executor/CpsNcmpTaskExecutorSpec.groovy
@@ -0,0 +1,86 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2023 Nordix Foundation
+ *  ================================================================================
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *  ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.rest.executor
+
+import ch.qos.logback.classic.Level
+import ch.qos.logback.classic.Logger
+import ch.qos.logback.classic.spi.ILoggingEvent
+import ch.qos.logback.core.read.ListAppender
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.BeforeEach
+import org.onap.cps.notification.NotificationErrorHandler
+import org.slf4j.LoggerFactory
+import spock.lang.Specification
+
+class CpsNcmpTaskExecutorSpec extends Specification {
+
+    def objectUnderTest = new CpsNcmpTaskExecutor()
+    def logger = Spy(ListAppender<ILoggingEvent>)
+    def enoughTime = 100
+
+    @BeforeEach
+    void setup() {
+        ((Logger) LoggerFactory.getLogger(CpsNcmpTaskExecutor.class)).addAppender(logger);
+        logger.start();
+    }
+
+    @AfterEach
+    void teardown() {
+        ((Logger) LoggerFactory.getLogger(NotificationErrorHandler.class)).detachAndStopAllAppenders();
+    }
+
+    def 'Execute successful task.'() {
+        when: 'task is executed'
+            objectUnderTest.executeTask(taskSupplier(), enoughTime)
+        and: 'wait a little for async execution completion'
+            Thread.sleep(10)
+        then: 'an event is logged with level INFO'
+            def loggingEvent = getLoggingEvent()
+            assert loggingEvent.level == Level.INFO
+        and: 'the log indicates the task completed successfully'
+            assert loggingEvent.formattedMessage == 'Async task completed successfully.'
+    }
+
+    def 'Execute failing task.'() {
+        when: 'task is executed'
+            objectUnderTest.executeTask(taskSupplierForFailingTask(), enoughTime)
+        and: 'wait a little for async execution completion'
+            Thread.sleep(10)
+        then: 'an event is logged with level ERROR'
+            def loggingEvent = getLoggingEvent()
+            assert loggingEvent.level == Level.ERROR
+        and: 'the original error message is logged'
+            assert loggingEvent.formattedMessage.contains('original exception message')
+    }
+
+    def taskSupplier() {
+        return () -> 'hello world'
+    }
+
+    def taskSupplierForFailingTask() {
+        return () -> { throw new RuntimeException('original exception message') }
+    }
+
+    def getLoggingEvent() {
+        return logger.list[0]
+    }
+
+}