Outbox Pattern

@Service
@Slf4j
public class OrderService {

    private final OrderRepository orderRepo;
    private final OutboxRepository outboxRepo;

    @Transactional  // one database transaction covers both writes
    public Order createOrder(CreateOrderRequest req) {
        // 1. Business write
        Order order = orderRepo.save(Order.from(req));

        // 2. Outbox row in the SAME transaction (insert only: Debezium does not allow updates)
        OutboxEvent outboxEvent = OutboxEvent.builder()
            .aggregateId(order.getId())
            .aggregateType("Order")
            .type("OrderCreated")
            .payload(objectMapper.writeValueAsString(new OrderCreatedEvent(order)))
            .build();
        outboxRepo.save(outboxEvent);

        return order;
        // Debezium reads the outbox insert from the database log and publishes it to Kafka
    }
}

// Outbox entity: column names match the event router's defaults
@Entity
@Table(name = "outbox")
public class OutboxEvent {
    @Id @GeneratedValue private UUID id;
    @Column(name = "aggregateid") private String aggregateId;
    @Column(name = "aggregatetype") private String aggregateType;
    private String type;
    @Column(columnDefinition = "TEXT") private String payload;
}

{
  "name": "outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.dbname": "orders_db",
    "table.include.list": "public.outbox",
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "transforms.outbox.table.field.event.id": "id",
    "transforms.outbox.table.field.event.key": "aggregateid",
    "transforms.outbox.table.field.event.payload": "payload",
    "transforms.outbox.route.by.field": "aggregatetype",
    "transforms.outbox.route.topic.replacement": "orders.${routedByValue}",
    "transforms.outbox.table.fields.additional.placement": "type:header:eventType"
  }
}