diff options
Diffstat (limited to 'src')
22 files changed, 1005 insertions, 37 deletions
diff --git a/src/main/java/org/onap/cps/temporal/controller/event/listener/exception/EventListenerException.java b/src/main/java/org/onap/cps/temporal/controller/event/listener/exception/EventListenerException.java new file mode 100644 index 0000000..a9d1ce2 --- /dev/null +++ b/src/main/java/org/onap/cps/temporal/controller/event/listener/exception/EventListenerException.java @@ -0,0 +1,34 @@ +/* + * ============LICENSE_START======================================================= + * Copyright (c) 2021 Bell Canada. + * ================================================================================ + * 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.cps.temporal.controller.event.listener.exception; + +/** + * Class representing a listener exception related to system event error. + */ +public class EventListenerException extends RuntimeException { + + public EventListenerException(final String message) { + super(message); + } + + public EventListenerException(final String message, final Throwable cause) { + super(message, cause); + } + +} diff --git a/src/main/java/org/onap/cps/temporal/controller/event/listener/exception/InvalidEventEnvelopException.java b/src/main/java/org/onap/cps/temporal/controller/event/listener/exception/InvalidEventEnvelopException.java new file mode 100644 index 0000000..df4e756 --- /dev/null +++ b/src/main/java/org/onap/cps/temporal/controller/event/listener/exception/InvalidEventEnvelopException.java @@ -0,0 +1,73 @@ +/* + * ============LICENSE_START======================================================= + * Copyright (c) 2021 Bell Canada. + * ================================================================================ + * 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.cps.temporal.controller.event.listener.exception; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.List; +import lombok.AllArgsConstructor; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.ToString; + +/** + * Class representing an invalid event envelop exception. + */ +@Getter +public class InvalidEventEnvelopException extends EventListenerException { + + private final List<InvalidField> invalidFields = new ArrayList<>(); + + public InvalidEventEnvelopException(final String message) { + super(message); + } + + public void addInvalidField(final InvalidField invalidField) { + this.invalidFields.add(invalidField); + } + + public boolean hasInvalidFields() { + return ! this.invalidFields.isEmpty(); + } + + @Override + public String getMessage() { + return String.format("%s. invalidFields: %s", super.getMessage(), this.invalidFields.toString()); + } + + @AllArgsConstructor + @Getter + @EqualsAndHashCode + @ToString + public static class InvalidField implements Serializable { + + private static final long serialVersionUID = -7118283787669377391L; + + private final ErrorType errorType; + private final String fieldName; + private final String actualValue; + private final String expectedValue; + + public enum ErrorType { + UNEXPECTED, MISSING + } + + } + +} diff --git a/src/main/java/org/onap/cps/temporal/controller/event/listener/kafka/DataUpdatedEventListener.java b/src/main/java/org/onap/cps/temporal/controller/event/listener/kafka/DataUpdatedEventListener.java new file mode 100644 index 0000000..79c9d92 --- /dev/null +++ b/src/main/java/org/onap/cps/temporal/controller/event/listener/kafka/DataUpdatedEventListener.java @@ -0,0 +1,129 @@ +/* + * ============LICENSE_START======================================================= + * Copyright (c) 2021 Bell Canada. + * ================================================================================ + * 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.cps.temporal.controller.event.listener.kafka; + +import static org.onap.cps.temporal.controller.event.listener.exception.InvalidEventEnvelopException.InvalidField.ErrorType.MISSING; +import static org.onap.cps.temporal.controller.event.listener.exception.InvalidEventEnvelopException.InvalidField.ErrorType.UNEXPECTED; + +import java.net.URI; +import java.net.URISyntaxException; +import lombok.extern.slf4j.Slf4j; +import org.onap.cps.event.model.CpsDataUpdatedEvent; +import org.onap.cps.temporal.controller.event.listener.exception.EventListenerException; +import org.onap.cps.temporal.controller.event.listener.exception.InvalidEventEnvelopException; +import org.onap.cps.temporal.controller.event.model.CpsDataUpdatedEventMapper; +import org.onap.cps.temporal.service.NetworkDataService; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.stereotype.Component; +import org.springframework.util.StringUtils; + +/** + * Listener for data updated events. + */ +@Component +@Slf4j +public class DataUpdatedEventListener { + + private static final URI EVENT_SOURCE; + + static { + try { + EVENT_SOURCE = new URI("urn:cps:org.onap.cps"); + } catch (final URISyntaxException e) { + throw new EventListenerException("Invalid URI for event source.", e); + } + } + + private static final String EVENT_TYPE = "org.onap.cps.data-updated-event"; + + private final NetworkDataService networkDataService; + private final CpsDataUpdatedEventMapper cpsDataUpdatedEventMapper; + + /** + * Constructor. + */ + public DataUpdatedEventListener( + final NetworkDataService networkDataService, final CpsDataUpdatedEventMapper cpsDataUpdatedEventMapper) { + this.networkDataService = networkDataService; + this.cpsDataUpdatedEventMapper = cpsDataUpdatedEventMapper; + } + + /** + * Consume the specified event. + * + * @param cpsDataUpdatedEvent the data updated event to be consumed and persisted. + */ + @KafkaListener(topics = "${app.listener.data-updated.topic}", errorHandler = "dataUpdatedEventListenerErrorHandler") + public void consume(final CpsDataUpdatedEvent cpsDataUpdatedEvent) { + + log.debug("Receiving {} ...", cpsDataUpdatedEvent); + + // Validate event envelop + validateEventEnvelop(cpsDataUpdatedEvent); + + // Map event to entity + final var networkData = this.cpsDataUpdatedEventMapper.eventToEntity(cpsDataUpdatedEvent); + log.debug("Persisting {} ...", networkData); + + // Persist entity + final var persistedNetworkData = this.networkDataService.addNetworkData(networkData); + log.debug("Persisted {}", persistedNetworkData); + + } + + private void validateEventEnvelop(final CpsDataUpdatedEvent cpsDataUpdatedEvent) { + + final var invalidEventEnvelopException = new InvalidEventEnvelopException("Validation failure"); + + // Schema + if (cpsDataUpdatedEvent.getSchema() == null) { + invalidEventEnvelopException.addInvalidField( + new InvalidEventEnvelopException.InvalidField( + MISSING, "schema", null, + CpsDataUpdatedEvent.Schema.URN_CPS_ORG_ONAP_CPS_DATA_UPDATED_EVENT_SCHEMA_1_1_0_SNAPSHOT + .value())); + } + // Id + if (!StringUtils.hasText(cpsDataUpdatedEvent.getId())) { + invalidEventEnvelopException.addInvalidField( + new InvalidEventEnvelopException.InvalidField( + MISSING, "id", null, null)); + } + // Source + if (cpsDataUpdatedEvent.getSource() == null || !cpsDataUpdatedEvent.getSource().equals(EVENT_SOURCE)) { + invalidEventEnvelopException.addInvalidField( + new InvalidEventEnvelopException.InvalidField( + UNEXPECTED, "source", + cpsDataUpdatedEvent.getSource() != null + ? cpsDataUpdatedEvent.getSource().toString() : null, EVENT_SOURCE.toString())); + } + // Type + if (cpsDataUpdatedEvent.getType() == null || !cpsDataUpdatedEvent.getType().equals(EVENT_TYPE)) { + invalidEventEnvelopException.addInvalidField( + new InvalidEventEnvelopException.InvalidField( + UNEXPECTED, "type", cpsDataUpdatedEvent.getType(), EVENT_TYPE)); + } + + if (invalidEventEnvelopException.hasInvalidFields()) { + throw invalidEventEnvelopException; + } + + } + +} diff --git a/src/main/java/org/onap/cps/temporal/controller/event/listener/kafka/DataUpdatedEventListenerErrorHandler.java b/src/main/java/org/onap/cps/temporal/controller/event/listener/kafka/DataUpdatedEventListenerErrorHandler.java new file mode 100644 index 0000000..7a4ee7f --- /dev/null +++ b/src/main/java/org/onap/cps/temporal/controller/event/listener/kafka/DataUpdatedEventListenerErrorHandler.java @@ -0,0 +1,44 @@ +/* + * ============LICENSE_START======================================================= + * Copyright (c) 2021 Bell Canada. + * ================================================================================ + * 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.cps.temporal.controller.event.listener.kafka; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.kafka.listener.KafkaListenerErrorHandler; +import org.springframework.kafka.listener.ListenerExecutionFailedException; +import org.springframework.messaging.Message; +import org.springframework.stereotype.Component; + +/** + * Class responsible to handle errors for data updated event listener. + */ +@Component +@Slf4j +class DataUpdatedEventListenerErrorHandler implements KafkaListenerErrorHandler { + + @Override + public Object handleError(final Message<?> message, final ListenerExecutionFailedException exception) { + log.error( + "Failed to process message {}. Error cause is {}.", + message, + exception.getCause() != null ? exception.getCause().toString() : null, + exception); + return exception; + } + +} diff --git a/src/main/java/org/onap/cps/temporal/controller/event/model/CpsDataUpdatedEventMapper.java b/src/main/java/org/onap/cps/temporal/controller/event/model/CpsDataUpdatedEventMapper.java new file mode 100644 index 0000000..9ef25d5 --- /dev/null +++ b/src/main/java/org/onap/cps/temporal/controller/event/model/CpsDataUpdatedEventMapper.java @@ -0,0 +1,56 @@ +/* + * ============LICENSE_START======================================================= + * Copyright (c) 2021 Bell Canada. + * ================================================================================ + * 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.cps.temporal.controller.event.model; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.time.OffsetDateTime; +import java.time.format.DateTimeFormatter; +import org.mapstruct.Mapper; +import org.mapstruct.Mapping; +import org.onap.cps.event.model.CpsDataUpdatedEvent; +import org.onap.cps.event.model.Data; +import org.onap.cps.temporal.domain.NetworkData; + +/** + * Mapper for data updated event schema. + */ +@Mapper(componentModel = "spring") +public abstract class CpsDataUpdatedEventMapper { + + private static final DateTimeFormatter ISO_TIMESTAMP_FORMATTER = + DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSZ"); + + @Mapping(source = "content.observedTimestamp", target = "observedTimestamp") + @Mapping(source = "content.dataspaceName", target = "dataspace") + @Mapping(source = "content.schemaSetName", target = "schemaSet") + @Mapping(source = "content.anchorName", target = "anchor") + @Mapping(source = "content.data", target = "payload") + @Mapping(expression = "java(null)", target = "createdTimestamp") + public abstract NetworkData eventToEntity(CpsDataUpdatedEvent cpsDataUpdatedEvent); + + String map(final Data data) throws JsonProcessingException { + return data != null ? new ObjectMapper().writeValueAsString(data) : null; + } + + OffsetDateTime map(final String timestamp) { + return timestamp != null ? OffsetDateTime.parse(timestamp, ISO_TIMESTAMP_FORMATTER) : null; + } + +} diff --git a/src/main/java/org/onap/cps/temporal/controller/QueryController.java b/src/main/java/org/onap/cps/temporal/controller/web/QueryController.java index d083dc9..fae95cc 100644 --- a/src/main/java/org/onap/cps/temporal/controller/QueryController.java +++ b/src/main/java/org/onap/cps/temporal/controller/web/QueryController.java @@ -16,7 +16,7 @@ * ============LICENSE_END========================================================= */ -package org.onap.cps.temporal.controller; +package org.onap.cps.temporal.controller.web; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RestController; diff --git a/src/main/java/org/onap/cps/temporal/domain/NetworkData.java b/src/main/java/org/onap/cps/temporal/domain/NetworkData.java index aa2ce95..1537e4a 100644 --- a/src/main/java/org/onap/cps/temporal/domain/NetworkData.java +++ b/src/main/java/org/onap/cps/temporal/domain/NetworkData.java @@ -48,27 +48,30 @@ import org.hibernate.annotations.TypeDef; @TypeDef(name = "jsonb", typeClass = JsonBinaryType.class) public class NetworkData implements Serializable { - private static final long serialVersionUID = -8032810412816532433L; + private static final long serialVersionUID = 8886477871334560919L; @Id + @NotNull @Column private OffsetDateTime observedTimestamp; @Id + @NotNull @Column private String dataspace; @Id + @NotNull @Column private String anchor; @NotNull - @Column + @Column(updatable = false) private String schemaSet; @NotNull @Type(type = "jsonb") - @Column(columnDefinition = "jsonb") + @Column(columnDefinition = "jsonb", updatable = false) private String payload; @CreationTimestamp diff --git a/src/main/java/org/onap/cps/temporal/domain/NetworkDataId.java b/src/main/java/org/onap/cps/temporal/domain/NetworkDataId.java index e9742e2..18c4dcf 100644 --- a/src/main/java/org/onap/cps/temporal/domain/NetworkDataId.java +++ b/src/main/java/org/onap/cps/temporal/domain/NetworkDataId.java @@ -32,10 +32,10 @@ import lombok.NoArgsConstructor; @EqualsAndHashCode public class NetworkDataId implements Serializable { - private static final long serialVersionUID = 1L; + private static final long serialVersionUID = -1039604338648260766L; + private OffsetDateTime observedTimestamp; private String dataspace; private String anchor; - private OffsetDateTime observedTimestamp; -}
\ No newline at end of file +} diff --git a/src/main/java/org/onap/cps/temporal/service/NetworkDataServiceImpl.java b/src/main/java/org/onap/cps/temporal/service/NetworkDataServiceImpl.java index 2e7afb2..687ba85 100644 --- a/src/main/java/org/onap/cps/temporal/service/NetworkDataServiceImpl.java +++ b/src/main/java/org/onap/cps/temporal/service/NetworkDataServiceImpl.java @@ -18,24 +18,39 @@ package org.onap.cps.temporal.service; +import java.util.Optional; import lombok.extern.slf4j.Slf4j; import org.onap.cps.temporal.domain.NetworkData; +import org.onap.cps.temporal.domain.NetworkDataId; import org.onap.cps.temporal.repository.NetworkDataRepository; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.stereotype.Component; +import org.springframework.stereotype.Service; /** * Service implementation for Network Data. */ -@Component +@Service @Slf4j public class NetworkDataServiceImpl implements NetworkDataService { - @Autowired - NetworkDataRepository networkDataRepository; + private final NetworkDataRepository networkDataRepository; + + public NetworkDataServiceImpl(final NetworkDataRepository networkDataRepository) { + this.networkDataRepository = networkDataRepository; + } @Override public NetworkData addNetworkData(final NetworkData networkData) { - return networkDataRepository.save(networkData); + final var savedNetworkData = networkDataRepository.save(networkData); + if (savedNetworkData.getCreatedTimestamp() == null) { + // Data already exists and can not be inserted + final var id = + new NetworkDataId( + networkData.getObservedTimestamp(), networkData.getDataspace(), networkData.getAnchor()); + final Optional<NetworkData> existingNetworkData = networkDataRepository.findById(id); + throw new ServiceException( + "Failed to create network data. It already exists: " + (existingNetworkData.orElse(null))); + } + return savedNetworkData; } + } diff --git a/src/main/java/org/onap/cps/temporal/service/ServiceException.java b/src/main/java/org/onap/cps/temporal/service/ServiceException.java new file mode 100644 index 0000000..b9d7184 --- /dev/null +++ b/src/main/java/org/onap/cps/temporal/service/ServiceException.java @@ -0,0 +1,34 @@ +/* + * ============LICENSE_START======================================================= + * Copyright (c) 2021 Bell Canada. + * ================================================================================ + * 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.cps.temporal.service; + +/** + * Class representing a service exception related to business error. + */ +public class ServiceException extends RuntimeException { + + /** + * Instantiate a service exception with the specified message. + * @param message the exception message + */ + public ServiceException(final String message) { + super(message); + } + +} diff --git a/src/main/resources/application-sasl-ssl-kafka.yml b/src/main/resources/application-sasl-ssl-kafka.yml new file mode 100644 index 0000000..fdc6458 --- /dev/null +++ b/src/main/resources/application-sasl-ssl-kafka.yml @@ -0,0 +1,31 @@ +# ============LICENSE_START======================================================= +# Copyright (c) 2021 Bell Canada. +# ================================================================================ +# 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. +# ============LICENSE_END========================================================= + +# Spring profile configuration for sasl ssl Kafka + +spring: + kafka: + bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVER} + security: + protocol: SASL_SSL + ssl: + trust-store-type: JKS + trust-store-location: ${KAFKA_SSL_TRUST_STORE_LOCATION} + trust-store-password: ${KAFKA_SSL_TRUST_STORE_PASSWORD} + properties: + sasl.mechanism: SCRAM-SHA-512 + sasl.jaas.config: ${KAFKA_SASL_JAAS_CONFIG} + ssl.endpoint.identification.algorithm: diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index d4c799c..5fe30b0 100755 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -27,10 +27,26 @@ spring: change-log: classpath:/db/changelog/changelog-master.xml jpa: properties: - hibernate: - dialect: org.hibernate.dialect.PostgreSQLDialect + hibernate.dialect: org.hibernate.dialect.PostgreSQLDialect + hibernate.format_sql: true + hibernate.generate_statistics: false + kafka: + bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVER} + security: + protocol: PLAINTEXT + consumer: + group-id: ${KAFKA_CONSUMER_GROUP_ID:cps-temporal-group} + # Configures the Spring Kafka ErrorHandlingDeserializer that delegates to the 'real' deserializers + # See https://docs.spring.io/spring-kafka/docs/2.5.11.RELEASE/reference/html/#error-handling-deserializer + # and https://www.confluent.io/blog/spring-kafka-can-your-kafka-consumers-handle-a-poison-pill/ + key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer + value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer + properties: + spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer + spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer + spring.json.value.default.type: org.onap.cps.event.model.CpsDataUpdatedEvent -logging: - level: - org: - springframework: INFO +app: + listener: + data-updated: + topic: ${CPS_CHANGE_EVENT_TOPIC:cps.cfg-state-events} diff --git a/src/main/resources/logback.xml b/src/main/resources/logback.xml new file mode 100644 index 0000000..a75b7aa --- /dev/null +++ b/src/main/resources/logback.xml @@ -0,0 +1,43 @@ +<!-- + ============LICENSE_START======================================================= + Copyright (c) 2021 Bell Canada. + ================================================================================ + 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. + ============LICENSE_END========================================================= +--> + +<configuration> + + <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender"> + <encoder> + <pattern>%d - %highlight(%-5level) [%-20.20thread] %cyan(%logger{36}) - %msg%n</pattern> + </encoder> + </appender> + + <!-- Logger for cps classes --> + <logger name="org.onap.cps" level="info"/> + + <!-- Logger for sql statements. Set to info to disable, debug to enable --> + <logger name="org.hibernate.SQL" level="info"/> + + <!-- Logger for sql bindings. Set to info to disable, to trace to enable --> + <logger name="org.hibernate.type.descriptor.sql.BasicBinder" level="info"/> + + <!-- Logger for hibernate statistics. Set to warn to disable, to info to enable --> + <logger name="org.hibernate.engine.internal.StatisticalLoggingSessionEventListener" level="warn"/> + + <root level="info"> + <appender-ref ref="STDOUT" /> + </root> + +</configuration> diff --git a/src/test/groovy/org/onap/cps/temporal/controller/event/listener/kafka/DataUpdatedEventListenerIntegrationSpec.groovy b/src/test/groovy/org/onap/cps/temporal/controller/event/listener/kafka/DataUpdatedEventListenerIntegrationSpec.groovy new file mode 100644 index 0000000..4c362ad --- /dev/null +++ b/src/test/groovy/org/onap/cps/temporal/controller/event/listener/kafka/DataUpdatedEventListenerIntegrationSpec.groovy @@ -0,0 +1,124 @@ +/* + * ============LICENSE_START======================================================= + * Copyright (c) 2021 Bell Canada. + * ================================================================================ + * 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. + * ============LICENSE_END========================================================= +*/ + +package org.onap.cps.temporal.controller.event.listener.kafka + +import groovy.util.logging.Slf4j +import org.onap.cps.event.model.CpsDataUpdatedEvent +import org.onap.cps.temporal.repository.containers.TimescaleContainer +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.beans.factory.annotation.Value +import org.springframework.boot.test.context.SpringBootTest +import org.springframework.jdbc.core.JdbcTemplate +import org.springframework.kafka.core.KafkaTemplate +import org.springframework.test.context.DynamicPropertyRegistry +import org.springframework.test.context.DynamicPropertySource +import org.testcontainers.containers.KafkaContainer +import spock.lang.Shared +import spock.lang.Specification +import spock.util.concurrent.PollingConditions + +import java.util.concurrent.TimeUnit + +/** + * Integration test specification for data updated event listener. + * This integration test is running database and kafka dependencies as docker containers. + */ +@SpringBootTest +@Slf4j +class DataUpdatedEventListenerIntegrationSpec extends Specification { + + @Shared + def databaseTestContainer = TimescaleContainer.getInstance() + + static kafkaTestContainer = new KafkaContainer() + static { + Runtime.getRuntime().addShutdownHook(new Thread(kafkaTestContainer::stop)) + } + + def setupSpec() { + databaseTestContainer.start() + kafkaTestContainer.start() + } + + @Autowired + KafkaTemplate<String, CpsDataUpdatedEvent> kafkaTemplate + + @Autowired + JdbcTemplate jdbcTemplate + + @Value('${app.listener.data-updated.topic}') + String topic + + // Define event data + def aTimestamp = EventFixtures.currentIsoTimestamp() + def aDataspace = 'my-dataspace' + def aSchemaSet = 'my-schema-set' + def anAnchor = 'my-anchor' + + // Define sql queries for data validation + def sqlCount = "select count(*) from network_data" + def sqlSelect = "select * from network_data" + def sqlWhereClause = + ' where observed_timestamp = to_timestamp(?, \'YYYY-MM-DD"T"HH24:MI:SS.USTZHTZM\') ' + + 'and dataspace = ? ' + + 'and schema_set = ? ' + + 'and anchor = ?' + def sqlCountWithConditions = sqlCount + sqlWhereClause + def sqlSelectWithConditions = sqlSelect + sqlWhereClause + + def 'Processing a valid event'() { + given: "no event has been proceeded" + def initialRecordsCount = + jdbcTemplate.queryForObject(sqlCountWithConditions, Integer.class, + aTimestamp, aDataspace, aSchemaSet, anAnchor) + assert (initialRecordsCount == 0) + when: 'an event is produced' + def event = + EventFixtures.buildEvent( + timestamp: aTimestamp, dataspace: aDataspace, schemaSet: aSchemaSet, anchor: anAnchor) + this.kafkaTemplate.send(topic, event) + then: 'the event is proceeded' + def pollingCondition = new PollingConditions(timeout: 10, initialDelay: 1, factor: 2) + pollingCondition.eventually { + def finalRecordsCount = + jdbcTemplate.queryForObject( + sqlCountWithConditions, Integer.class, aTimestamp, aDataspace, aSchemaSet, anAnchor) + assert (finalRecordsCount == 1) + } + Map<String, Object> result = + jdbcTemplate.queryForMap(sqlSelectWithConditions, aTimestamp, aDataspace, aSchemaSet, anAnchor) + log.debug("Data retrieved from db: {}", result) + } + + def 'Processing an invalid event'() { + given: 'the number of network data records if known' + def initialRecordsCount = jdbcTemplate.queryForObject(sqlCount, Integer.class) + when: 'an invalid event is produced' + this.kafkaTemplate.send(topic, (CpsDataUpdatedEvent) null) + then: 'the event is not proceeded and no more network data record is created' + TimeUnit.SECONDS.sleep(3) + assert (jdbcTemplate.queryForObject(sqlCount, Integer.class) == initialRecordsCount) + } + + @DynamicPropertySource + static void registerKafkaProperties(DynamicPropertyRegistry registry) { + registry.add("spring.kafka.bootstrap-servers", kafkaTestContainer::getBootstrapServers) + } + +} diff --git a/src/test/groovy/org/onap/cps/temporal/controller/event/listener/kafka/DataUpdatedEventListenerSpec.groovy b/src/test/groovy/org/onap/cps/temporal/controller/event/listener/kafka/DataUpdatedEventListenerSpec.groovy new file mode 100644 index 0000000..d3a407c --- /dev/null +++ b/src/test/groovy/org/onap/cps/temporal/controller/event/listener/kafka/DataUpdatedEventListenerSpec.groovy @@ -0,0 +1,117 @@ +/* + * ============LICENSE_START======================================================= + * Copyright (c) 2021 Bell Canada. + * ================================================================================ + * 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.cps.temporal.controller.event.listener.kafka + +import org.mapstruct.factory.Mappers +import org.onap.cps.event.model.CpsDataUpdatedEvent +import org.onap.cps.temporal.controller.event.listener.exception.InvalidEventEnvelopException +import org.onap.cps.temporal.controller.event.model.CpsDataUpdatedEventMapper +import org.onap.cps.temporal.service.NetworkDataService +import spock.lang.Specification + +import static org.onap.cps.temporal.controller.event.listener.exception.InvalidEventEnvelopException.InvalidField.ErrorType.MISSING +import static org.onap.cps.temporal.controller.event.listener.exception.InvalidEventEnvelopException.InvalidField.ErrorType.UNEXPECTED + +/** + * Test specification for data updated event listener. + */ +class DataUpdatedEventListenerSpec extends Specification { + + // Define event data + def anEventType = 'my-event-type' + def anEventSource = new URI('my-event-source') + def aTimestamp = EventFixtures.currentIsoTimestamp() + def aDataspace = 'my-dataspace' + def aSchemaSet = 'my-schema-set' + def anAnchor = 'my-anchor' + def aDataName = 'my-data-name' + def aDataValue = 'my-data-value' + + // Define service mock + def mockService = Mock(NetworkDataService) + + // Define mapper + def mapper = Mappers.getMapper(CpsDataUpdatedEventMapper.class) + + // Define listener under test + def objectUnderTest = new DataUpdatedEventListener(mockService, mapper) + + def 'Event message consumption'() { + when: 'an event is received' + def event = + EventFixtures.buildEvent( + timestamp: aTimestamp, dataspace: aDataspace, schemaSet: aSchemaSet, anchor: anAnchor, + dataName: aDataName, dataValue: aDataValue) + objectUnderTest.consume(event) + then: 'network data service is requested to persisted the data change' + 1 * mockService.addNetworkData( + { + it.getObservedTimestamp() == EventFixtures.toOffsetDateTime(aTimestamp) + && it.getDataspace() == aDataspace + && it.getSchemaSet() == aSchemaSet + && it.getAnchor() == anAnchor + && it.getPayload() == String.format('{"%s":"%s"}', aDataName, aDataValue) + && it.getCreatedTimestamp() == null + } + ) + } + + def 'Event message consumption fails because of missing envelop'() { + when: 'an event without envelop information is received' + def invalidEvent = new CpsDataUpdatedEvent().withSchema(null) + objectUnderTest.consume(invalidEvent) + then: 'an exception is thrown with 4 invalid fields' + def e = thrown(InvalidEventEnvelopException) + e.getInvalidFields().size() == 4 + e.getInvalidFields().contains( + new InvalidEventEnvelopException.InvalidField( + MISSING,"schema", null, + CpsDataUpdatedEvent.Schema.URN_CPS_ORG_ONAP_CPS_DATA_UPDATED_EVENT_SCHEMA_1_1_0_SNAPSHOT + .value())) + e.getInvalidFields().contains( + new InvalidEventEnvelopException.InvalidField( + MISSING, "id", null, null)) + e.getInvalidFields().contains( + new InvalidEventEnvelopException.InvalidField( + UNEXPECTED, "source", null, EventFixtures.defaultEventSource.toString())) + e.getInvalidFields().contains( + new InvalidEventEnvelopException.InvalidField( + UNEXPECTED, "type", null, EventFixtures.defaultEventType)) + e.getMessage().contains(e.getInvalidFields().toString()) + } + + def 'Event message consumption fails because of invalid envelop'() { + when: 'an event with an invalid envelop is received' + def invalidEvent = + new CpsDataUpdatedEvent() + .withId('my-id').withSource(anEventSource).withType(anEventType) + objectUnderTest.consume(invalidEvent) + then: 'an exception is thrown with 2 invalid fields' + def e = thrown(InvalidEventEnvelopException) + e.getInvalidFields().size() == 2 + e.getInvalidFields().contains( + new InvalidEventEnvelopException.InvalidField( + UNEXPECTED, "type", anEventType, EventFixtures.defaultEventType)) + e.getInvalidFields().contains( + new InvalidEventEnvelopException.InvalidField( + UNEXPECTED, "source", anEventSource.toString(), + EventFixtures.defaultEventSource.toString())) + } + +} diff --git a/src/test/groovy/org/onap/cps/temporal/controller/event/listener/kafka/EventFixtures.groovy b/src/test/groovy/org/onap/cps/temporal/controller/event/listener/kafka/EventFixtures.groovy new file mode 100644 index 0000000..44a28de --- /dev/null +++ b/src/test/groovy/org/onap/cps/temporal/controller/event/listener/kafka/EventFixtures.groovy @@ -0,0 +1,72 @@ +/* + * ============LICENSE_START======================================================= + * Copyright (c) 2021 Bell Canada. + * ================================================================================ + * 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.cps.temporal.controller.event.listener.kafka + +import org.onap.cps.event.model.Content +import org.onap.cps.event.model.CpsDataUpdatedEvent +import org.onap.cps.event.model.Data + +import java.time.OffsetDateTime +import java.time.format.DateTimeFormatter + +/** + * This class contains utility fixtures methods for building and manipulating event data. + */ +class EventFixtures { + + static DateTimeFormatter isoTimestampFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSZ") + static String defaultEventType = 'org.onap.cps.data-updated-event' + static URI defaultEventSource = new URI('urn:cps:org.onap.cps') + + static CpsDataUpdatedEvent buildEvent(final Map map) { + CpsDataUpdatedEvent event = + new CpsDataUpdatedEvent() + .withId( + map.id != null ? map.id.toString() : UUID.randomUUID().toString()) + .withType( + map.eventType != null ? map.eventType.toString() : defaultEventType) + .withSource( + map.eventSource != null ? new URI(map.eventSource.toString()) : defaultEventSource) + .withContent( + new Content() + .withObservedTimestamp( + map.timestamp != null ? map.timestamp.toString() : currentTimestamp()) + .withDataspaceName( + map.dataspace != null ? map.dataspace.toString() : 'a-dataspace') + .withSchemaSetName( + map.schemaSet != null ? map.schemaSet.toString() : 'a-schema-set') + .withAnchorName( + map.anchor != null ? map.anchor.toString() : 'an-anchor') + .withData( + new Data().withAdditionalProperty( + map.dataName != null ? map.dataName.toString() : 'a-data-name', + map.dataValue != null ? map.dataValue : 'a-data-value'))) + + return event + } + + static String currentIsoTimestamp() { + return isoTimestampFormatter.format(OffsetDateTime.now()) + } + + static OffsetDateTime toOffsetDateTime(String timestamp) { + return OffsetDateTime.parse(timestamp, isoTimestampFormatter) + } + +} diff --git a/src/test/groovy/org/onap/cps/temporal/controller/event/model/CpsDataUpdatedEventMapperSpec.groovy b/src/test/groovy/org/onap/cps/temporal/controller/event/model/CpsDataUpdatedEventMapperSpec.groovy new file mode 100644 index 0000000..132ff6d --- /dev/null +++ b/src/test/groovy/org/onap/cps/temporal/controller/event/model/CpsDataUpdatedEventMapperSpec.groovy @@ -0,0 +1,131 @@ +/* + * ============LICENSE_START======================================================= + * Copyright (c) 2021 Bell Canada. + * ================================================================================ + * 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. + * ============LICENSE_END========================================================= + */ + +package org.onap.cps.temporal.controller.event.model + +import com.fasterxml.jackson.core.JsonProcessingException +import org.mapstruct.factory.Mappers +import org.onap.cps.event.model.Content +import org.onap.cps.event.model.CpsDataUpdatedEvent +import org.onap.cps.event.model.Data +import org.onap.cps.temporal.domain.NetworkData +import spock.lang.Specification + +import java.time.OffsetDateTime +import java.time.format.DateTimeFormatter + +/** + * Test specification for data updated event mapper. + */ +class CpsDataUpdatedEventMapperSpec extends Specification { + + def objectUnderTest = Mappers.getMapper(CpsDataUpdatedEventMapper.class); + + def 'Mapping a null event'() { + given: 'a null event' + def event = null + when: 'the event is mapped to an entity' + NetworkData result = objectUnderTest.eventToEntity(event) + then: 'the result entity is null' + result == null + } + + def 'Mapping an event whose properties are null'() { + given: 'an event whose properties are null' + def event = new CpsDataUpdatedEvent() + when: 'the event is mapped to an entity' + NetworkData result = objectUnderTest.eventToEntity(event) + then: 'the result entity is not null' + result != null + and: 'all result entity properties are null' + assertEntityPropertiesAreNull(result) + } + + def 'Mapping an event whose content properties are null'() { + given: 'an event whose content properties are null' + def event = new CpsDataUpdatedEvent().withContent(new Content()) + when: 'the event is mapped to an entity' + NetworkData result = objectUnderTest.eventToEntity(event) + then: 'the result entity is not null' + result != null + and: 'all result entity properties are null' + assertEntityPropertiesAreNull(result) + } + + def 'Mapping an event whose content data is empty'() { + given: 'an event whose content data is empty' + def event = new CpsDataUpdatedEvent().withContent(new Content().withData(new Data())) + when: 'the event is mapped to an entity' + NetworkData result = objectUnderTest.eventToEntity(event) + then: 'the result entity is not null' + result != null + and: 'the result entity payload is an empty json ' + result.getPayload() == "{}" + } + + def 'Mapping an event whose content data is invalid'() { + given: 'an event whose content data is invalid' + def event = + new CpsDataUpdatedEvent().withContent(new Content().withData( + new Data().withAdditionalProperty(null, null))) + when: 'the event is mapped to an entity' + NetworkData result = objectUnderTest.eventToEntity(event) + then: 'an runtime exception is thrown' + def e = thrown(RuntimeException) + e.getCause() instanceof JsonProcessingException + } + + def 'Mapping a valid complete event'() { + given: 'a valid complete event' + def isoTimestampFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSZ") + def aDataName = 'a-data-name' + def aDataValue = 'a-data-value' + def event = + new CpsDataUpdatedEvent() + .withContent( + new Content() + .withObservedTimestamp(isoTimestampFormatter.format(OffsetDateTime.now())) + .withDataspaceName('a-dataspace') + .withSchemaSetName('a-schema-set') + .withAnchorName('an-anchor') + .withData(new Data().withAdditionalProperty(aDataName, aDataValue))) + when: 'the event is mapped to an entity' + NetworkData result = objectUnderTest.eventToEntity(event) + then: 'the result entity is not null' + result != null + and: 'all result entity properties are the ones from the event' + result.getObservedTimestamp() == + OffsetDateTime.parse(event.getContent().getObservedTimestamp(), isoTimestampFormatter) + result.getDataspace() == event.getContent().getDataspaceName() + result.getSchemaSet() == event.getContent().getSchemaSetName() + result.getAnchor() == event.getContent().getAnchorName() + result.getPayload().contains(aDataValue) + result.getPayload().contains(aDataValue) + result.getCreatedTimestamp() == null + } + + private void assertEntityPropertiesAreNull(NetworkData networkData) { + assert networkData.getObservedTimestamp() == null + assert networkData.getDataspace() == null + assert networkData.getSchemaSet() == null + assert networkData.getAnchor() == null + assert networkData.getPayload() == null + assert networkData.getCreatedTimestamp() == null + } + +} diff --git a/src/test/groovy/org/onap/cps/temporal/controller/QueryControllerSpec.groovy b/src/test/groovy/org/onap/cps/temporal/controller/web/QueryControllerSpec.groovy index f718bf4..c834f38 100644 --- a/src/test/groovy/org/onap/cps/temporal/controller/QueryControllerSpec.groovy +++ b/src/test/groovy/org/onap/cps/temporal/controller/web/QueryControllerSpec.groovy @@ -16,7 +16,7 @@ * ============LICENSE_END========================================================= */ -package org.onap.cps.temporal.controller +package org.onap.cps.temporal.controller.web import spock.lang.Specification @@ -34,4 +34,4 @@ class QueryControllerSpec extends Specification { ! response.empty } -}
\ No newline at end of file +} diff --git a/src/test/groovy/org/onap/cps/temporal/repository/NetworkDataRepositorySpec.groovy b/src/test/groovy/org/onap/cps/temporal/repository/NetworkDataRepositorySpec.groovy index ec976ee..f66b35e 100644 --- a/src/test/groovy/org/onap/cps/temporal/repository/NetworkDataRepositorySpec.groovy +++ b/src/test/groovy/org/onap/cps/temporal/repository/NetworkDataRepositorySpec.groovy @@ -29,6 +29,9 @@ import spock.lang.Specification import java.time.OffsetDateTime +/** + * Test specification for network data repository. + */ @SpringBootTest @Testcontainers class NetworkDataRepositorySpec extends Specification { diff --git a/src/test/groovy/org/onap/cps/temporal/service/NetworkDataServiceImplSpec.groovy b/src/test/groovy/org/onap/cps/temporal/service/NetworkDataServiceImplSpec.groovy index 70ac2bc..9847f54 100644 --- a/src/test/groovy/org/onap/cps/temporal/service/NetworkDataServiceImplSpec.groovy +++ b/src/test/groovy/org/onap/cps/temporal/service/NetworkDataServiceImplSpec.groovy @@ -18,28 +18,51 @@ package org.onap.cps.temporal.service +import org.onap.cps.temporal.domain.NetworkDataId + import java.time.OffsetDateTime import org.onap.cps.temporal.domain.NetworkData import org.onap.cps.temporal.repository.NetworkDataRepository import spock.lang.Specification +/** + * Test specification for network data service. + */ class NetworkDataServiceImplSpec extends Specification { - def objectUnderTest = new NetworkDataServiceImpl() - def mockNetworkDataRepository = Mock(NetworkDataRepository) + def objectUnderTest = new NetworkDataServiceImpl(mockNetworkDataRepository) + def networkData = new NetworkData() - def setup() { - objectUnderTest.networkDataRepository = mockNetworkDataRepository + def 'Add network data successfully.'() { + given: 'network data repository is persisting network data it is asked to save' + def persistedNetworkData = new NetworkData() + persistedNetworkData.setCreatedTimestamp(OffsetDateTime.now()) + mockNetworkDataRepository.save(networkData) >> persistedNetworkData + when: 'a new network data is added' + def result = objectUnderTest.addNetworkData(networkData) + then: 'result network data is the one that has been persisted' + result == persistedNetworkData + result.getCreatedTimestamp() != null + networkData.getCreatedTimestamp() == null } - def 'Add network data in timeseries database.'() { + def 'Add network data fails because already added'() { + given: 'network data repository is not able to create data it is asked to persist ' + + 'and reveals it with null created timestamp on network data entity' + def persistedNetworkData = new NetworkData() + persistedNetworkData.setCreatedTimestamp(null) + mockNetworkDataRepository.save(networkData) >> persistedNetworkData + and: 'existing data can be retrieved' + def existing = new NetworkData() + existing.setCreatedTimestamp(OffsetDateTime.now().minusYears(1)) + mockNetworkDataRepository.findById(_ as NetworkDataId) >> Optional.of(existing) when: 'a new network data is added' objectUnderTest.addNetworkData(networkData) - then: ' repository service is called with the correct parameters' - 1 * mockNetworkDataRepository.save(networkData) + then: 'network service exception is thrown' + thrown(ServiceException) } } diff --git a/src/test/java/org/onap/cps/temporal/architecture/LayeredArchitectureTest.java b/src/test/java/org/onap/cps/temporal/architecture/LayeredArchitectureTest.java index d47e8a5..a70e914 100644 --- a/src/test/java/org/onap/cps/temporal/architecture/LayeredArchitectureTest.java +++ b/src/test/java/org/onap/cps/temporal/architecture/LayeredArchitectureTest.java @@ -70,4 +70,4 @@ public class LayeredArchitectureTest { .should().onlyHaveDependentClassesThat() .resideInAnyPackage(SERVICE_PACKAGE, REPOSITORY_PACKAGE); -}
\ No newline at end of file +} diff --git a/src/test/resources/application.yml b/src/test/resources/application.yml index afaff6c..3ac13a9 100644 --- a/src/test/resources/application.yml +++ b/src/test/resources/application.yml @@ -20,17 +20,37 @@ server: spring: datasource: url: ${DB_URL} - password: ${DB_PASSWORD} username: ${DB_USERNAME} + password: ${DB_PASSWORD} liquibase: change-log: classpath:/db/changelog/changelog-master.xml jpa: - open-in-view: false properties: - hibernate: - dialect: org.hibernate.dialect.PostgreSQLDialect + hibernate.dialect: org.hibernate.dialect.PostgreSQLDialect + hibernate.format_sql: true + hibernate.generate_statistics: false + kafka: + bootstrap-servers: localhost:9092 + security: + protocol: PLAINTEXT + consumer: + group-id: cps-temporal-group + # Configures the Spring Kafka ErrorHandlingDeserializer that delegates to the 'real' deserializers + # See https://docs.spring.io/spring-kafka/docs/2.5.11.RELEASE/reference/html/#error-handling-deserializer + # and https://www.confluent.io/blog/spring-kafka-can-your-kafka-consumers-handle-a-poison-pill/ + key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer + value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer + auto-offset-reset: earliest + properties: + spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer + spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer + spring.json.value.default.type: org.onap.cps.event.model.CpsDataUpdatedEvent + # Following is not cps-temporal configuration. It is configuration for the producer used for integration tests + producer: + key-serializer: org.apache.kafka.common.serialization.StringSerializer + value-serializer: org.springframework.kafka.support.serializer.JsonSerializer -logging: - level: - org: - springframework: INFO +app: + listener: + data-updated: + topic: cps.cfg-state-events |