5.5 Define Events

Define the events used.

In the previous section we have configured the tables for our event handling. Now it’s time to define our actual events for the application.

Event overview

We will build the following event flow. Keep in mind that messages are not directly written to Kafka.

Environment

Define Events

To define our events we have to implement ExportedEvent. This event implementation is responsible to fill up our outboxevent table. The definition of the values is as follows:

Field Description
id unique id. Can be used by consumers to detect duplicates. Should be generated when creating the event Object.
aggregatetype the origin of a given event. It could be the microservice name or for example in case of domain-driven design the origin domain. For example “order” or “stock”.
aggregateid contains the id affected by a given event. In case of the aggregatetype order this could be the order id. This id is used as the key for Kafka messages. Only messages with the same key are guaranteed to be processed in order as they are placed in the same partition of a kafka topic.
type the actual type of the event. For example “OrderCreated” or “StockComplete”. We will use this information to trigger specific event handlers on the consumer side.
payload in our example a JSON structure which contains the event body. For example the actual Order with Article id and Amount ordered.

Task 5.5.1 - Implement OrderCreated Event

Given the information above. Try to implement the OrderCreated event.

  • Complete implementation ch.puzzle.mm.debezium.event.entity.OrderCreatedEvent
    • Set the required fields accordingly (AGGREGATE_TYPE and EVENT_TYPE)

For help, here’s an excerpt of the order outboxevent table. For simplicity the timestamp is omitted.

1
2
3
4
5
6
7
admin=# select id, aggregatetype, aggregateid, type, payload from outboxevent;
    id     | aggregatetype | aggregateid |      type      |                         payload
-----------+---------------+-------------+----------------+---------------------------------------------------------
 bed8bd... | Order         | 100000      | OrderCreated   | {"orderId":100000,"items":[{"articleId":1,"amount":9}]}
 28e777... | Order         | 100002      | OrderCreated   | {"orderId":100002,"items":[{"articleId":1,"amount":3}]}
 e09acc... | Order         | 100000      | OrderCancelled | {"orderId":100000,"items":[{"articleId":1,"amount":9}]}
 3709f2... | Order         | 100004      | OrderCreated   | {"orderId":100004,"items":[{"articleId":1,"amount":3}]}
Task Hint
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
public class OrderCreatedEvent implements ExportedEvent<String, JsonNode> {

    private static ObjectMapper mapper = new ObjectMapper();

    private static final String AGGREGATE_TYPE = "Order";
    private static final String EVENT_TYPE = "OrderCreated";

    private final UUID id;
    private final Long orderId;
    private final JsonNode jsonNode;
    private final Instant timestamp;

    public OrderCreatedEvent(Instant created, ShopOrder shopOrder) {
        this.id = UUID.randomUUID();
        this.orderId = shopOrder.id;
        this.timestamp = created;
        this.jsonNode = asJson(shopOrder);
    }

    @Override
    public String getAggregateId() {
        return String.valueOf(orderId);
    }

    @Override
    public String getAggregateType() {
        return AGGREGATE_TYPE;
    }

    @Override
    public String getType() {
        return EVENT_TYPE;
    }

    @Override
    public Instant getTimestamp() {
        return timestamp;
    }

    @Override
    public JsonNode getPayload() {
        return jsonNode;
    }

    public ObjectNode asJson(ShopOrder order) {
        ObjectNode asJson = mapper.createObjectNode()
                .put("orderId", order.id);

        ArrayNode items = asJson.putArray("items");

        for (ArticleOrder article : order.getArticleOrders()) {
            items.add(mapper.createObjectNode()
                    .put("articleId", article.getArticleId())
                    .put("amount", article.getAmount())
            );
        }

        return asJson;
    }
}