summaryrefslogtreecommitdiffstats
path: root/cps-ncmp-service
diff options
context:
space:
mode:
Diffstat (limited to 'cps-ncmp-service')
-rw-r--r--cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/client/DmiRestClient.java2
-rw-r--r--cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/EventsPublisher.java56
-rw-r--r--cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/executor/AsyncTaskExecutor.java2
-rw-r--r--cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/CmHandleQueryApiParameters.java2
-rw-r--r--cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/CmHandleQueryServiceParameters.java2
-rw-r--r--cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/CmSubscriptionEvent.java4
-rw-r--r--cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/CmSubscriptionStatus.java2
-rw-r--r--cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/ConditionApiProperties.java2
-rw-r--r--cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/DataOperationDefinition.java2
-rw-r--r--cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/DataOperationRequest.java2
-rw-r--r--cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/EventPublisherSpec.groovy3
-rw-r--r--cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/kafka/ConsumerBaseSpec.groovy2
12 files changed, 35 insertions, 46 deletions
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/client/DmiRestClient.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/client/DmiRestClient.java
index 6a8310c207..7066f80165 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/client/DmiRestClient.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/client/DmiRestClient.java
@@ -60,7 +60,7 @@ public class DmiRestClient {
} catch (final HttpStatusCodeException httpStatusCodeException) {
final String exceptionMessage = "Unable to " + operationType.toString() + " resource data.";
throw new HttpClientRequestException(exceptionMessage, httpStatusCodeException.getResponseBodyAsString(),
- httpStatusCodeException.getStatusCode().value());
+ httpStatusCodeException.getRawStatusCode());
}
}
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/EventsPublisher.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/EventsPublisher.java
index 58d0d2d48f..4c90648521 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/EventsPublisher.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/EventsPublisher.java
@@ -22,7 +22,6 @@ package org.onap.cps.ncmp.api.impl.events;
import io.cloudevents.CloudEvent;
import java.util.Map;
-import java.util.concurrent.CompletableFuture;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
@@ -32,6 +31,8 @@ import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.SerializationUtils;
+import org.springframework.util.concurrent.ListenableFuture;
+import org.springframework.util.concurrent.ListenableFutureCallback;
/**
* EventsPublisher to publish events.
@@ -60,17 +61,9 @@ public class EventsPublisher<T> {
* @param event message payload
*/
public void publishCloudEvent(final String topicName, final String eventKey, final CloudEvent event) {
- final CompletableFuture<SendResult<String, CloudEvent>> eventFuture =
- cloudEventKafkaTemplate.send(topicName, eventKey, event);
- eventFuture.whenComplete((result, e) -> {
- if (e == null) {
- log.debug("Successfully published event to topic : {} , Event : {}",
- result.getRecordMetadata().topic(), result.getProducerRecord().value());
-
- } else {
- log.error("Unable to publish event to topic : {} due to {}", topicName, e.getMessage());
- }
- });
+ final ListenableFuture<SendResult<String, CloudEvent>> eventFuture
+ = cloudEventKafkaTemplate.send(topicName, eventKey, event);
+ eventFuture.addCallback(handleCallback(topicName));
}
/**
@@ -83,16 +76,9 @@ public class EventsPublisher<T> {
*/
@Deprecated(forRemoval = true)
public void publishEvent(final String topicName, final String eventKey, final T event) {
- final CompletableFuture<SendResult<String, T>> eventFuture =
- legacyKafkaEventTemplate.send(topicName, eventKey, event);
- eventFuture.whenComplete((result, e) -> {
- if (e == null) {
- log.debug("Successfully published event to topic : {} , Event : {}",
- result.getRecordMetadata().topic(), result.getProducerRecord().value());
- } else {
- log.error("Unable to publish event to topic : {} due to {}", topicName, e.getMessage());
- }
- });
+ final ListenableFuture<SendResult<String, T>> eventFuture
+ = legacyKafkaEventTemplate.send(topicName, eventKey, event);
+ eventFuture.addCallback(handleCallback(topicName));
}
/**
@@ -107,16 +93,8 @@ public class EventsPublisher<T> {
final ProducerRecord<String, T> producerRecord =
new ProducerRecord<>(topicName, null, eventKey, event, eventHeaders);
- final CompletableFuture<SendResult<String, T>> eventFuture =
- legacyKafkaEventTemplate.send(producerRecord);
- eventFuture.whenComplete((result, ex) -> {
- if (ex != null) {
- log.error("Unable to publish event to topic : {} due to {}", topicName, ex.getMessage());
- } else {
- log.debug("Successfully published event to topic : {} , Event : {}",
- result.getRecordMetadata().topic(), result.getProducerRecord().value());
- }
- });
+ final ListenableFuture<SendResult<String, T>> eventFuture = legacyKafkaEventTemplate.send(producerRecord);
+ eventFuture.addCallback(handleCallback(topicName));
}
/**
@@ -133,6 +111,20 @@ public class EventsPublisher<T> {
publishEvent(topicName, eventKey, convertToKafkaHeaders(eventHeaders), event);
}
+ private ListenableFutureCallback<SendResult<String, ?>> handleCallback(final String topicName) {
+ return new ListenableFutureCallback<>() {
+ @Override
+ public void onFailure(final Throwable throwable) {
+ log.error("Unable to publish event to topic : {} due to {}", topicName, throwable.getMessage());
+ }
+
+ @Override
+ public void onSuccess(final SendResult<String, ?> sendResult) {
+ log.debug("Successfully published event to topic : {} , Event : {}",
+ sendResult.getRecordMetadata().topic(), sendResult.getProducerRecord().value());
+ }
+ };
+ }
private Headers convertToKafkaHeaders(final Map<String, Object> eventMessageHeaders) {
final Headers eventHeaders = new RecordHeaders();
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/executor/AsyncTaskExecutor.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/executor/AsyncTaskExecutor.java
index 3d08185e14..cac78af050 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/executor/AsyncTaskExecutor.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/executor/AsyncTaskExecutor.java
@@ -22,12 +22,12 @@ package org.onap.cps.ncmp.api.inventory.sync.executor;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
-import jakarta.annotation.PostConstruct;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;
import java.util.function.Supplier;
+import javax.annotation.PostConstruct;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/CmHandleQueryApiParameters.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/CmHandleQueryApiParameters.java
index dd8dcd60ac..bf6600d97e 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/CmHandleQueryApiParameters.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/CmHandleQueryApiParameters.java
@@ -24,9 +24,9 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonInclude.Include;
import com.fasterxml.jackson.annotation.JsonProperty;
-import jakarta.validation.Valid;
import java.util.Collections;
import java.util.List;
+import javax.validation.Valid;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.Setter;
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/CmHandleQueryServiceParameters.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/CmHandleQueryServiceParameters.java
index 5eeafaca33..774f04b5a4 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/CmHandleQueryServiceParameters.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/CmHandleQueryServiceParameters.java
@@ -23,9 +23,9 @@ package org.onap.cps.ncmp.api.models;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonInclude.Include;
import com.fasterxml.jackson.annotation.JsonProperty;
-import jakarta.validation.Valid;
import java.util.Collections;
import java.util.List;
+import javax.validation.Valid;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.Setter;
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/CmSubscriptionEvent.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/CmSubscriptionEvent.java
index c20696a8b6..2a60b2ab16 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/CmSubscriptionEvent.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/CmSubscriptionEvent.java
@@ -22,10 +22,10 @@ package org.onap.cps.ncmp.api.models;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonProperty;
-import jakarta.validation.Valid;
-import jakarta.validation.constraints.NotNull;
import java.util.ArrayList;
import java.util.List;
+import javax.validation.Valid;
+import javax.validation.constraints.NotNull;
import lombok.Getter;
import lombok.Setter;
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/CmSubscriptionStatus.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/CmSubscriptionStatus.java
index 5541a01bd8..bba560785d 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/CmSubscriptionStatus.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/CmSubscriptionStatus.java
@@ -22,7 +22,7 @@ package org.onap.cps.ncmp.api.models;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonProperty;
-import jakarta.validation.constraints.NotNull;
+import javax.validation.constraints.NotNull;
import lombok.Getter;
import lombok.Setter;
import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionStatus;
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/ConditionApiProperties.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/ConditionApiProperties.java
index 5cb2ed376c..9f6d64e164 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/ConditionApiProperties.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/ConditionApiProperties.java
@@ -22,10 +22,10 @@ package org.onap.cps.ncmp.api.models;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonProperty;
-import jakarta.validation.Valid;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import javax.validation.Valid;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.Setter;
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/DataOperationDefinition.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/DataOperationDefinition.java
index 366d845832..8182fbfcc8 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/DataOperationDefinition.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/DataOperationDefinition.java
@@ -23,9 +23,9 @@ package org.onap.cps.ncmp.api.models;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonProperty;
-import jakarta.validation.Valid;
import java.util.ArrayList;
import java.util.List;
+import javax.validation.Valid;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.Setter;
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/DataOperationRequest.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/DataOperationRequest.java
index a4d070c38d..6fa7d5c755 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/DataOperationRequest.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/models/DataOperationRequest.java
@@ -23,9 +23,9 @@ package org.onap.cps.ncmp.api.models;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonProperty;
-import jakarta.validation.Valid;
import java.util.Collections;
import java.util.List;
+import javax.validation.Valid;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.Setter;
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/EventPublisherSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/EventPublisherSpec.groovy
index d0f1afd5d5..59a43caf9e 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/EventPublisherSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/EventPublisherSpec.groovy
@@ -29,7 +29,6 @@ import org.apache.kafka.common.TopicPartition
import org.onap.cps.ncmp.init.SubscriptionModelLoader
import org.slf4j.LoggerFactory
import org.springframework.kafka.support.SendResult
-import spock.lang.Ignore
import spock.lang.Specification
class EventPublisherSpec extends Specification {
@@ -49,7 +48,6 @@ class EventPublisherSpec extends Specification {
((Logger) LoggerFactory.getLogger(SubscriptionModelLoader.class)).detachAndStopAllAppenders()
}
- @Ignore
def 'Callback handling on success.'() {
given: 'a send result'
def producerRecord = new ProducerRecord('topic-1', 'my value')
@@ -68,7 +66,6 @@ class EventPublisherSpec extends Specification {
}
- @Ignore
def 'Callback handling on failure.'() {
when: 'the callback handler processes a failure'
def callbackHandler = objectUnderTest.handleCallback('my topic')
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/kafka/ConsumerBaseSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/kafka/ConsumerBaseSpec.groovy
index 28f8b02880..940c59d57f 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/kafka/ConsumerBaseSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/kafka/ConsumerBaseSpec.groovy
@@ -25,7 +25,7 @@ import org.springframework.boot.test.context.SpringBootTest
import org.springframework.kafka.config.KafkaListenerEndpointRegistry
import org.springframework.kafka.test.utils.ContainerTestUtils
-@SpringBootTest(classes = KafkaListenerEndpointRegistry.class)
+@SpringBootTest
class ConsumerBaseSpec extends MessagingBaseSpec {
@Autowired