Then, use `kafkacat` tool to produce a data updated event into the Kafka topic:
```bash
-docker run -i --rm --network=host edenhill/kafkacat:1.6.0 -b localhost:19092 -t cps.cfg-state-events -D/ -P <<EOF
+docker run -i --rm --network=host edenhill/kafkacat:1.6.0 -b localhost:19092 -t cps.data-updated-events -D/ -P <<EOF
{
- "schema": "urn:cps:org.onap.cps:data-updated-event-schema:1.1.0-SNAPSHOT",
+ "schema": "urn:cps:org.onap.cps:data-updated-event-schema:v1",
"id": "38aa6cc6-264d-4ede-b534-18f5c1f403ea",
"source": "urn:cps:org.onap.cps",
"type": "org.onap.cps.data-updated-event",
<dependency>
<groupId>org.onap.cps</groupId>
<artifactId>cps-events</artifactId>
- <version>1.1.0</version>
+ <version>1.1.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springdoc</groupId>
@Slf4j
public class DataUpdatedEventListener {
+ private static final String EVENT_SCHEMA_URN_PREFIX = "urn:cps:org.onap.cps:data-updated-event-schema:v";
private static final URI EVENT_SOURCE;
static {
new InvalidEventEnvelopException("Validation failure", cpsDataUpdatedEvent);
// Validate schema
- if (cpsDataUpdatedEvent.getSchema() == null) {
+ if (cpsDataUpdatedEvent.getSchema() == null
+ || !cpsDataUpdatedEvent.getSchema().toString().startsWith(EVENT_SCHEMA_URN_PREFIX)) {
invalidEventEnvelopException.addInvalidField(
new InvalidEventEnvelopException.InvalidField(
- MISSING, "schema", null,
- CpsDataUpdatedEvent.Schema.URN_CPS_ORG_ONAP_CPS_DATA_UPDATED_EVENT_SCHEMA_1_1_0_SNAPSHOT
- .value()));
+ UNEXPECTED, "schema",
+ cpsDataUpdatedEvent.getSchema() != null ? cpsDataUpdatedEvent.getSchema().toString() : null,
+ EVENT_SCHEMA_URN_PREFIX + "99"));
}
// Validate id
if (!StringUtils.hasText(cpsDataUpdatedEvent.getId())) {
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
+ spring.json.use.type.headers: false
app:
listener:
data-updated:
- topic: ${CPS_CHANGE_EVENT_TOPIC:cps.cfg-state-events}
+ topic: ${CPS_CHANGE_EVENT_TOPIC:cps.data-updated-events}
query:
response:
max-page-size: 10000
*/
class DataUpdatedEventListenerSpec extends Specification {
+ public static final String EXPECTED_SCHEMA_EXCEPTION_MESSAGE = 'urn:cps:org.onap.cps:data-updated-event-schema:v99'
+
// Define event data
def anEventType = 'my-event-type'
+ def anEventSchema = new URI('my-event-schema')
def anEventSource = new URI('my-event-source')
def aTimestamp = EventFixtures.currentIsoTimestamp()
def aDataspace = 'my-dataspace'
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()))
+ UNEXPECTED,"schema", null, EXPECTED_SCHEMA_EXCEPTION_MESSAGE))
e.getInvalidFields().contains(
new InvalidEventEnvelopException.InvalidField(
MISSING, "id", null, null))
when: 'an event with an invalid envelop is received'
def invalidEvent =
new CpsDataUpdatedEvent()
- .withId('my-id').withSource(anEventSource).withType(anEventType)
+ .withId('my-id')
+ .withSchema(anEventSchema)
+ .withSource(anEventSource)
+ .withType(anEventType)
objectUnderTest.consume(invalidEvent)
then: 'an exception is thrown with 2 invalid fields'
def e = thrown(InvalidEventEnvelopException)
e.getCpsDataUpdatedEvent() == invalidEvent
- e.getInvalidFields().size() == 2
+ e.getInvalidFields().size() == 3
+ e.getInvalidFields().contains(
+ new InvalidEventEnvelopException.InvalidField(
+ UNEXPECTED, "schema", anEventSchema.toString(),
+ EXPECTED_SCHEMA_EXCEPTION_MESSAGE))
e.getInvalidFields().contains(
new InvalidEventEnvelopException.InvalidField(
UNEXPECTED, "type", anEventType, EventFixtures.defaultEventType))
static DateTimeFormatter isoTimestampFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSZ")
static String defaultEventType = 'org.onap.cps.data-updated-event'
+ static URI defaultEventSchema = new URI('urn:cps:org.onap.cps:data-updated-event-schema:v1')
static URI defaultEventSource = new URI('urn:cps:org.onap.cps')
static CpsDataUpdatedEvent buildEvent(final Map map) {
CpsDataUpdatedEvent event =
new CpsDataUpdatedEvent()
+ .withSchema(
+ map.eventSchema != null ? new URI(map.eventSchema.toString()) : defaultEventSchema)
.withId(
map.id != null ? map.id.toString() : UUID.randomUUID().toString())
.withType(
app:
listener:
data-updated:
- topic: cps.cfg-state-events
+ topic: cps.data-updated-events
query:
response:
max-page-size: 20