minimumIdle: 5
maximumPoolSize: 80
idleTimeout: 60000
- connectionTimeout: 120000
+ connectionTimeout: 30000
leakDetectionThreshold: 30000
pool-name: CpsDatabasePool
ncmp:
dmi:
httpclient:
- connectionTimeoutInSeconds: 180
+ connectionTimeoutInSeconds: 30
maximumConnectionsPerRoute: 50
maximumConnectionsTotal: 100
idleConnectionEvictionThresholdInSeconds: 5
import java.util.Map;
import java.util.UUID;
+import java.util.function.BiConsumer;
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.impl.utils.data.operation.ResourceDataOperationRequestUtils;
import org.onap.cps.ncmp.api.models.CmResourceAddress;
import org.onap.cps.ncmp.api.models.DataOperationRequest;
import org.onap.cps.ncmp.rest.exceptions.OperationNotSupportedException;
final DataOperationRequest dataOperationRequest,
final String authorization) {
final String requestId = UUID.randomUUID().toString();
- cpsNcmpTaskExecutor.executeTask(
+ cpsNcmpTaskExecutor.executeTaskWithErrorHandling(
getTaskSupplierForDataOperationRequest(topicParamInQuery, dataOperationRequest, requestId, authorization),
+ getTaskCompletionHandlerForDataOperationRequest(topicParamInQuery, dataOperationRequest, requestId),
timeOutInMilliSeconds);
return ResponseEntity.ok(Map.of("requestId", requestId));
}
};
}
+ private static BiConsumer<Object, Throwable> getTaskCompletionHandlerForDataOperationRequest(
+ final String topicParamInQuery,
+ final DataOperationRequest dataOperationRequest,
+ final String requestId) {
+ return (result, throwable) ->
+ ResourceDataOperationRequestUtils.handleAsyncTaskCompletionForDataOperationsRequest(topicParamInQuery,
+ requestId, dataOperationRequest, throwable);
+ }
+
}
/*
* ============LICENSE_START=======================================================
- * Copyright (C) 2022-2023 Nordix Foundation
+ * Copyright (C) 2022-2024 Nordix Foundation
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
import java.util.function.Supplier;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
@Service
public class CpsNcmpTaskExecutor {
+ /**
+ * Execute a task asynchronously, and invoke completion handler when done.
+ *
+ * @param taskSupplier functional method is get() task needed to be executed asynchronously
+ * @param taskCompletionHandler the action to perform on task completion or error
+ * @param timeOutInMillis the time-out value in milliseconds
+ */
+ public void executeTaskWithErrorHandling(final Supplier<Object> taskSupplier,
+ final BiConsumer<Object, Throwable> taskCompletionHandler,
+ final long timeOutInMillis) {
+ CompletableFuture.supplyAsync(taskSupplier)
+ .orTimeout(timeOutInMillis, MILLISECONDS)
+ .whenCompleteAsync(taskCompletionHandler);
+ }
+
/**
* Execute a task asynchronously.
*
- * @param taskSupplier functional method is get() task need to executed asynchronously
+ * @param taskSupplier functional method is get() task needed to be executed asynchronously
* @param timeOutInMillis the time-out value in milliseconds
*/
public void executeTask(final Supplier<Object> taskSupplier, final long timeOutInMillis) {
- CompletableFuture.supplyAsync(taskSupplier::get)
- .orTimeout(timeOutInMillis, MILLISECONDS)
- .whenCompleteAsync((taskResult, throwable) -> handleTaskCompletion(throwable));
+ executeTaskWithErrorHandling(taskSupplier, (taskResult, throwable) -> handleTaskCompletion(throwable),
+ timeOutInMillis);
}
private void handleTaskCompletion(final Throwable throwable) {
and: 'async request id is generated'
assert response.contentAsString.contains('requestId')
then: 'the request is handled asynchronously'
- 1 * mockCpsTaskExecutor.executeTask(*_)
+ 1 * mockCpsTaskExecutor.executeTaskWithErrorHandling(*_)
where: 'the following data stores are used'
datastore << [PASSTHROUGH_RUNNING, PASSTHROUGH_OPERATIONAL]
}
when: 'data operation request is executed'
objectUnderTest.executeRequest('someTopic', new DataOperationRequest(), NO_AUTH_HEADER)
then: 'the task is executed in an async fashion or not'
- expectedCalls * spiedCpsNcmpTaskExecutor.executeTask(*_)
+ expectedCalls * spiedCpsNcmpTaskExecutor.executeTaskWithErrorHandling(*_)
where: 'the following parameters are used'
scenario | notificationFeatureEnabled || expectedCalls
'on' | true || 1
when: 'data operation request is executed'
objectUnderTest.executeRequest('myTopic', dataOperationRequest, NO_AUTH_HEADER)
then: 'the task is executed in an async fashion'
- 1 * spiedCpsNcmpTaskExecutor.executeTask(*_)
+ 1 * spiedCpsNcmpTaskExecutor.executeTaskWithErrorHandling(*_)
and: 'the network service is invoked'
new PollingConditions().within(1) {
assert networkServiceMethodCalled == true
def objectUnderTest = new CpsNcmpTaskExecutor()
def logger = Spy(ListAppender<ILoggingEvent>)
def enoughTime = 100
+ def notEnoughTime = 10
void setup() {
((Logger) LoggerFactory.getLogger(CpsNcmpTaskExecutor.class)).addAppender(logger)
assert loggingEvent.formattedMessage.contains('original exception message')
}
+ def 'Task times out.'() {
+ when: 'task is executed without enough time to complete'
+ objectUnderTest.executeTask(taskSupplierForLongRunningTask(), notEnoughTime)
+ then: 'an event is logged with level ERROR'
+ new PollingConditions().within(1) {
+ def loggingEvent = getLoggingEvent()
+ assert loggingEvent.level == Level.ERROR
+ }
+ and: 'a timeout error message is logged'
+ assert loggingEvent.formattedMessage.contains('java.util.concurrent.TimeoutException')
+ }
+
def taskSupplier() {
return () -> 'hello world'
}
return () -> { throw new RuntimeException('original exception message') }
}
+ def taskSupplierForLongRunningTask() {
+ return () -> { sleep(enoughTime) }
+ }
+
def getLoggingEvent() {
return logger.list[0]
}
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
DmiDataOperation.buildDmiDataOperationRequestBodyWithoutCmHandles(dataOperationDefinitionIn),
CM_HANDLES_NOT_READY, nonReadyCmHandleIds);
}
- if (!cmHandleIdsPerResponseCodesPerOperation.isEmpty()) {
- publishErrorMessageToClientTopic(topicParamInQuery, requestId, cmHandleIdsPerResponseCodesPerOperation);
- }
+ publishErrorMessageToClientTopic(topicParamInQuery, requestId, cmHandleIdsPerResponseCodesPerOperation);
return dmiDataOperationsOutPerDmiServiceName;
}
+ /**
+ * Handles the async task completion for an entire data, publishing errors to client topic on task failure.
+ *
+ * @param topicParamInQuery client given topic
+ * @param requestId unique identifier per request
+ * @param dataOperationRequest incoming data operation request details
+ * @param throwable error cause, or null if task completed with no exception
+ */
+ public static void handleAsyncTaskCompletionForDataOperationsRequest(
+ final String topicParamInQuery,
+ final String requestId,
+ final DataOperationRequest dataOperationRequest,
+ final Throwable throwable) {
+ if (throwable == null) {
+ log.info("Data operations request {} completed.", requestId);
+ } else if (throwable instanceof TimeoutException) {
+ log.error("Data operations request {} timed out.", requestId);
+ ResourceDataOperationRequestUtils.publishErrorMessageToClientTopicForEntireOperation(topicParamInQuery,
+ requestId, dataOperationRequest, NcmpResponseStatus.DMI_SERVICE_NOT_RESPONDING);
+ } else {
+ log.error("Data operations request {} failed.", requestId, throwable);
+ ResourceDataOperationRequestUtils.publishErrorMessageToClientTopicForEntireOperation(topicParamInQuery,
+ requestId, dataOperationRequest, NcmpResponseStatus.UNKNOWN_ERROR);
+ }
+ }
+
+ /**
+ * Creates data operation cloud event for when the entire data operation fails and publishes it to client topic.
+ *
+ * @param topicParamInQuery client given topic
+ * @param requestId unique identifier per request
+ * @param dataOperationRequestIn incoming data operation request details
+ * @param ncmpResponseStatus response code to be sent for all cm handle ids in all operations
+ */
+ private static void publishErrorMessageToClientTopicForEntireOperation(
+ final String topicParamInQuery,
+ final String requestId,
+ final DataOperationRequest dataOperationRequestIn,
+ final NcmpResponseStatus ncmpResponseStatus) {
+
+ final MultiValueMap<DmiDataOperation, Map<NcmpResponseStatus, List<String>>>
+ cmHandleIdsPerResponseCodesPerOperation = new LinkedMultiValueMap<>();
+
+ for (final DataOperationDefinition dataOperationDefinitionIn :
+ dataOperationRequestIn.getDataOperationDefinitions()) {
+ cmHandleIdsPerResponseCodesPerOperation.add(
+ DmiDataOperation.buildDmiDataOperationRequestBodyWithoutCmHandles(dataOperationDefinitionIn),
+ Map.of(ncmpResponseStatus, dataOperationDefinitionIn.getCmHandleIds()));
+ }
+ publishErrorMessageToClientTopic(topicParamInQuery, requestId, cmHandleIdsPerResponseCodesPerOperation);
+ }
+
/**
* Creates data operation cloud event and publish it to client topic.
*
* @param clientTopic client given topic
* @param requestId unique identifier per request
- * @param cmHandleIdsPerResponseCodesPerOperation list of cm handle ids per operation with response code
+ * @param cmHandleIdsPerResponseCodesPerOperation list of cm handle ids per operation with response code
*/
public static void publishErrorMessageToClientTopic(final String clientTopic,
final String requestId,
final MultiValueMap<DmiDataOperation,
Map<NcmpResponseStatus, List<String>>>
cmHandleIdsPerResponseCodesPerOperation) {
- final CloudEvent dataOperationCloudEvent = DataOperationEventCreator.createDataOperationEvent(clientTopic,
- requestId, cmHandleIdsPerResponseCodesPerOperation);
- final EventsPublisher<CloudEvent> eventsPublisher = CpsApplicationContext.getCpsBean(EventsPublisher.class);
- eventsPublisher.publishCloudEvent(clientTopic, requestId, dataOperationCloudEvent);
+ if (!cmHandleIdsPerResponseCodesPerOperation.isEmpty()) {
+ final CloudEvent dataOperationCloudEvent = DataOperationEventCreator.createDataOperationEvent(clientTopic,
+ requestId, cmHandleIdsPerResponseCodesPerOperation);
+ final EventsPublisher<CloudEvent> eventsPublisher = CpsApplicationContext.getCpsBean(EventsPublisher.class);
+ eventsPublisher.publishCloudEvent(clientTopic, requestId, dataOperationCloudEvent);
+ }
}
private static Map<String, String> getDmiServiceNamesPerCmHandleId(
import io.cloudevents.kafka.impl.KafkaHeaders
import org.apache.kafka.clients.consumer.KafkaConsumer
import org.onap.cps.events.EventsPublisher
+import org.onap.cps.ncmp.api.NcmpResponseStatus
import org.onap.cps.ncmp.api.impl.utils.context.CpsApplicationContext
import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle
import org.onap.cps.ncmp.api.impl.inventory.CmHandleState
import org.spockframework.spring.SpringBean
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.test.context.ContextConfiguration
+
import java.time.Duration
+import java.util.concurrent.TimeoutException
import static org.onap.cps.ncmp.api.impl.events.mapper.CloudEventMapper.toTargetEvent
@ContextConfiguration(classes = [EventsPublisher, CpsApplicationContext, ObjectMapper])
class ResourceDataOperationRequestUtilsSpec extends MessagingBaseSpec {
- def cloudEventKafkaConsumer = new KafkaConsumer<>(eventConsumerConfigProperties('test', CloudEventDeserializer))
def static clientTopic = 'my-topic-name'
def static dataOperationType = 'org.onap.cps.ncmp.events.async1_0_0.DataOperationEvent'
def 'Process per data operation request with non-ready, non-existing cm handle and publish event to client specified topic'() {
given: 'consumer subscribing to client topic'
+ def cloudEventKafkaConsumer = new KafkaConsumer<>(eventConsumerConfigProperties('test-1', CloudEventDeserializer))
cloudEventKafkaConsumer.subscribe([clientTopic])
and: 'data operation request having non-ready and non-existing cm handle ids'
def dataOperationRequestJsonData = TestUtils.getResourceFileContent('dataOperationRequest.json')
when: 'data operation request is processed'
ResourceDataOperationRequestUtils.processPerDefinitionInDataOperationsRequest(clientTopic, 'request-id', dataOperationRequest, yangModelCmHandles)
and: 'subscribed client specified topic is polled and first record is selected'
- def consumerRecordOut = cloudEventKafkaConsumer.poll(Duration.ofMillis(1500))[0]
+ def consumerRecordOut = cloudEventKafkaConsumer.poll(Duration.ofMillis(1500)).last()
then: 'verify cloud compliant headers'
def consumerRecordOutHeaders = consumerRecordOut.headers()
assert KafkaHeaders.getParsedKafkaHeader(consumerRecordOutHeaders, 'ce_id') != null
and: 'data operation response event response size is 3'
dataOperationResponseEvent.data.responses.size() == 3
and: 'verify published data operation response as json string'
- def dataOperationResponseEventJson = TestUtils.getResourceFileContent('dataOperationResponseEvent.json')
+ def dataOperationResponseEventJson = TestUtils.getResourceFileContent('dataOperationResponseEvent.json')
jsonObjectMapper.asJsonString(dataOperationResponseEvent.data.responses) == dataOperationResponseEventJson
}
+ def 'Publish error response for entire data operations request when async task fails'() {
+ given: 'consumer subscribing to client topic'
+ def cloudEventKafkaConsumer = new KafkaConsumer<>(eventConsumerConfigProperties(consumerGroupId, CloudEventDeserializer))
+ cloudEventKafkaConsumer.subscribe([clientTopic])
+ and: 'data operation request having non-ready and non-existing cm handle ids'
+ def dataOperationRequestJsonData = TestUtils.getResourceFileContent('dataOperationRequest.json')
+ def dataOperationRequest = jsonObjectMapper.convertJsonString(dataOperationRequestJsonData, DataOperationRequest.class)
+ when: 'an error occurs for the entire data operations request'
+ ResourceDataOperationRequestUtils.handleAsyncTaskCompletionForDataOperationsRequest(clientTopic, 'request-id', dataOperationRequest, exceptionThrown)
+ and: 'subscribed client specified topic is polled and first record is selected'
+ def consumerRecordOut = cloudEventKafkaConsumer.poll(Duration.ofMillis(1500)).last()
+ def dataOperationResponseEvent = toTargetEvent(consumerRecordOut.value(), DataOperationEvent.class)
+ then: 'data operation response event response size is 3'
+ dataOperationResponseEvent.data.responses.size() == 3
+ and: 'all 3 have the expected error code'
+ dataOperationResponseEvent.data.responses.each {
+ assert it.statusCode == errorReportedToClientTopic.code
+ }
+ where:
+ scenario | exceptionThrown | consumerGroupId || errorReportedToClientTopic
+ 'task timed out' | new TimeoutException() | 'test-2' || NcmpResponseStatus.DMI_SERVICE_NOT_RESPONDING
+ 'unspecified error' | new RuntimeException() | 'test-3' || NcmpResponseStatus.UNKNOWN_ERROR
+ }
+
static def getYangModelCmHandles() {
def dmiProperties = [new YangModelCmHandle.Property('prop', 'some DMI property')]
def readyState = new CompositeStateBuilder().withCmHandleState(CmHandleState.READY).withLastUpdatedTimeNow().build()
+-----------------+------------------------------------------------------+-----------------------------------+
| 1 | successfully applied subscription | CM Data Notification Subscription |
+-----------------+------------------------------------------------------+-----------------------------------+
- | 100 | cm handle id(s) is(are) not found | Data Operation, Inventory |
+ | 100 | cm handle id(s) is(are) not found | All features |
+-----------------+------------------------------------------------------+-----------------------------------+
| 101 | cm handle(s) not ready | Data Operation |
+-----------------+------------------------------------------------------+-----------------------------------+
+-----------------+------------------------------------------------------+-----------------------------------+
| 107 | southbound system is busy | Data Operation |
+-----------------+------------------------------------------------------+-----------------------------------+
- | 108 | Unknown error | Inventory |
+ | 108 | Unknown error | All features |
+-----------------+------------------------------------------------------+-----------------------------------+
| 109 | cm-handle already exists | Inventory |
+-----------------+------------------------------------------------------+-----------------------------------+
| | |
+--------------------------------------+--------------------------------------------------------+
+Bug Fixes
+---------
+3.4.8
+ - `CPS-2186 <https://jira.onap.org/browse/CPS-2186>`_ Report async task failures to client topic during data operations request
+
Features
--------
minimumIdle: 5
maximumPoolSize: 80
idleTimeout: 60000
- connectionTimeout: 120000
+ connectionTimeout: 30000
leakDetectionThreshold: 30000
pool-name: CpsDatabasePool
queue-capacity: 500
wait-for-tasks-to-complete-on-shutdown: true
thread-name-prefix: Async-
- time-out-value-in-ms: 2000
+ time-out-value-in-ms: 60000
springdoc:
swagger-ui:
ncmp:
dmi:
httpclient:
- connectionTimeoutInSeconds: 180
+ connectionTimeoutInSeconds: 30
maximumConnectionsPerRoute: 50
maximumConnectionsTotal: 100
idleConnectionEvictionThresholdInSeconds: 5
Code Review / cps.git / commitdiff