hpgrahsl

{
  "before": {
    "dbserver1.outbox_demo.outbox_event.Value": {
      "id": "6c810398-4b25-42ca-b18a-7428e16fefae",
      "aggregate_id": "1",
      "aggregate_type": "com.github.hpgrahsl.ms.outbox.sample.model.PurchaseOrder",
      "payload": "{\"id\":1,\"customerId\":1234,\"orderDate\":\"2019-07-19T10:50:15.528\",\"lineItems\":[{\"id\":1,\"item\":\"ABC\",\"quantity\":12,\"totalPrice\":49.25,\"status\":\"ENTERED\"},{\"id\":2,\"item\":\"XYZ\",\"quantity\":98,\"totalPrice\":99.25,\"status\":\"ENTERED\"}],\"totalValue\":148.5}",
      "timestamp": 1563526215,
      "type": "com.github.hpgrahsl.ms.outbox.sample.event.OrderUpsertedEvent"
    }

hpgrahsl

{
  "before": null,
  "after": {
    "dbserver1.outbox_demo.outbox_event.Value": {
      "id": "6c810398-4b25-42ca-b18a-7428e16fefae",
      "aggregate_id": "1",
      "aggregate_type": "com.github.hpgrahsl.ms.outbox.sample.model.PurchaseOrder",
      "payload": "{\"id\":1,\"customerId\":1234,\"orderDate\":\"2019-07-19T10:50:15.528\",\"lineItems\":[{\"id\":1,\"item\":\"ABC\",\"quantity\":12,\"totalPrice\":49.25,\"status\":\"ENTERED\"},{\"id\":2,\"item\":\"XYZ\",\"quantity\":98,\"totalPrice\":99.25,\"status\":\"ENTERED\"}],\"totalValue\":148.5}",
      "timestamp": 1563526215,
      "type": "com.github.hpgrahsl.ms.outbox.sample.event.OrderUpsertedEvent"

hpgrahsl

{
  "name": "mdb-sink-outbox-raw",
  "config": {
    "key.converter":"io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url":"http://localhost:8081",
    "value.converter":"io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url":"http://localhost:8081",
    "connector.class": "at.grahsl.kafka.connect.mongodb.MongoDbSinkConnector",
    "topics": "dbserver1.outbox-demo.outbox_event",
    "mongodb.connection.uri": "mongodb://localhost:27017/outboxed",

hpgrahsl

bin/kafka-avro-console-consumer --bootstrap-server localhost:9092 --topic dbserver1.outbox-demo.outbox_event --from-beginning | jq

hpgrahsl

{
  "name": "mysql-outbox-src-connector-01",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",
    "database.hostname": "localhost",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.server.id": "12345",

hpgrahsl

@Service
public class OrderService {

    //... 

    @Transactional
    public PurchaseOrder updateOrderLineStatus(long orderId, long orderLineId, OrderLineStatus newStatus) {
        PurchaseOrder po = repository.findById(orderId)
                .orElseThrow(() -> new EntityNotFoundException("order with id " + orderId + " doesn't exist!"));
        OrderLineStatus oldStatus = po.updateOrderLine(orderLineId, newStatus);

hpgrahsl

@Service
public class OrderService {

  //...

  @Transactional
  public PurchaseOrder placeOrder(PurchaseOrder order) {
    repository.save(order); //NOTE: OrderUpsertedEvent automatically published behind the scenes
    return order;
  }

hpgrahsl

public class OrderUpsertedEvent implements Outboxable {

    private static ObjectMapper MAPPER = new ObjectMapper();

    private final Long id;
    private final JsonNode payload;
    private final Long timestamp;

    static {
        MAPPER.registerModule(new JavaTimeModule());

hpgrahsl

@Entity
public class PurchaseOrder {

    //...

    @DomainEvents
    private Collection<Outboxable> triggerOutboxEvents() {
        return Arrays.asList(OrderUpsertedEvent.of(this));
    }

hpgrahsl

@Component
public class OutboxListener {

    private OutboxEventRepository repository;

    public OutboxListener(OutboxEventRepository repository) {
        this.repository = repository;
    }

    @EventListener