Outbox Pattern với Debezium - Phần 2

Để làm ví dụ về việc hiện thực Outbox Pattern với Debezium, mình sẽ có một hệ thống với 2 service 1 và 2.

Service 1 sẽ expose một API để thêm mới thông tin sinh viên vào database.
Thông tin sinh viên mới được thêm này sẽ được capture bởi Debezium và được publish tới Service 2 thông qua Apache Kafka.

Chúng ta sẽ sử dụng Outbox Pattern để đảm bảo tính nhất quán của dữ liệu các bạn nhé!

High level architecture của hệ thống mình sẽ như sau:

Service 1

Service 1 là một Spring Boot application sử dụng Web Starter, Data JPA Starter, PostgreSQL driver, Lombok và Docker Compose Support như sau:

PostgreSQL database server, mình sẽ sử dụng tính năng của Spring Boot hỗ trợ cho Docker Compose để có thể chạy một PostgreSQL database khi chạy ứng dụng lên. Nội dung của tập tin Docker Compose như sau:

services:
  postgres:
    image: 'postgres:latest'
    volumes:
      - ./src/main/resources/db.sql:/docker-entrypoint-initdb.d/db.sql
    environment:
      - POSTGRES_DB=outbox_pattern_example
      - POSTGRES_PASSWORD=123456
      - POSTGRES_USER=khanh
    command:
      - postgres
      - -c
      - wal_level=logical
      - -c
      - shared_preload_libraries=pgoutput
    ports:
      - '5430:5432'

services:

postgres:

image: 'postgres:latest'

volumes:

- ./src/main/resources/db.sql:/docker-entrypoint-initdb.d/db.sql

environment:

- POSTGRES_DB=outbox_pattern_example

- POSTGRES_PASSWORD=123456

- POSTGRES_USER=khanh

command:

- postgres

- -c

- wal_level=logical

- -c

- shared_preload_libraries=pgoutput

ports:

- '5430:5432'

Trong tập tin db.sql ở thư mục src/main/resources, table student sẽ được mình định nghĩa như sau:

CREATE TABLE student (
    id   UUID PRIMARY KEY,
    name VARCHAR(100)
);

CREATE TABLE student (

id UUID PRIMARY KEY,

name VARCHAR(100)

);

Entity của class Student và class Repository của nó, có nội dung như sau:

package com.huongdanjava.systemdesign.outboxpattern.repository.entity;

import jakarta.persistence.Entity;
import jakarta.persistence.Id;
import java.util.UUID;
import lombok.Getter;
import lombok.Setter;

@Entity
@Getter
@Setter
public class Student {

  @Id
  private UUID id;
  private String name;

}

package com.huongdanjava.systemdesign.outboxpattern.repository.entity;

import jakarta.persistence.Entity;

import jakarta.persistence.Id;

import java.util.UUID;

import lombok.Getter;

import lombok.Setter;

@Entity

@Getter

@Setter

public class Student {

@Id

private UUID id;

private String name;

}

và:

package com.huongdanjava.systemdesign.outboxpattern.repository;

import com.huongdanjava.systemdesign.outboxpattern.repository.entity.Student;
import java.util.UUID;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.stereotype.Repository;

@Repository
public interface StudentRepository extends JpaRepository<Student, UUID> {

}

package com.huongdanjava.systemdesign.outboxpattern.repository;

import com.huongdanjava.systemdesign.outboxpattern.repository.entity.Student;

import java.util.UUID;

import org.springframework.data.jpa.repository.JpaRepository;

import org.springframework.stereotype.Repository;

@Repository

public interface StudentRepository extends JpaRepository<Student, UUID> {

}

Cũng trong tập tin db.sql này, mình cũng định nghĩa nội dung của outbox table như sau:

CREATE TABLE outbox (
    id           UUID PRIMARY KEY,
    aggregate_id UUID    NOT NULL,
    type         VARCHAR NOT NULL,
    event_type   VARCHAR NOT NULL,
    payload      VARCHAR   NOT NULL
);

CREATE TABLE outbox (

id UUID PRIMARY KEY,

aggregate_id UUID NOT NULL,

type VARCHAR NOT NULL,

event_type VARCHAR NOT NULL,

payload VARCHAR NOT NULL

);

Cột id dùng làm unique id cho mỗi message sẽ được gửi tới Apache Kafka.
aggregate_id là id của domain object liên quan đến event. Trong ví dụ của mình thì là id của sinh viên đó các bạn.
type là domain object model liên quan đến event, cho ví dụ của mình là student đó các bạn.
event_type là loại event liên quan đến domain object model. Type có thể là thêm mới, update, xoá, …
payload là nội dung dưới dạng JSON của event.

Mình cũng thêm entity và class Repository cho outbox table này như sau:

package com.huongdanjava.systemdesign.outboxpattern.repository.entity;

import jakarta.persistence.Entity;
import jakarta.persistence.Id;
import java.util.UUID;
import lombok.Getter;
import lombok.Setter;

@Entity
@Getter
@Setter
public class Outbox {

  @Id
  private UUID id;
  private UUID aggregateId;
  private String type;
  private String eventType;
  private String payload;
}

package com.huongdanjava.systemdesign.outboxpattern.repository.entity;

import jakarta.persistence.Entity;

import jakarta.persistence.Id;

import java.util.UUID;

import lombok.Getter;

import lombok.Setter;

@Entity

@Getter

@Setter

public class Outbox {

@Id

private UUID id;

private UUID aggregateId;

private String type;

private String eventType;

private String payload;

}

và:

package com.huongdanjava.systemdesign.outboxpattern.repository;

import com.huongdanjava.systemdesign.outboxpattern.repository.entity.Student;
import java.util.UUID;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.stereotype.Repository;

@Repository
public interface StudentRepository extends JpaRepository<Student, UUID> {

}

package com.huongdanjava.systemdesign.outboxpattern.repository;

import com.huongdanjava.systemdesign.outboxpattern.repository.entity.Student;

import java.util.UUID;

import org.springframework.data.jpa.repository.JpaRepository;

import org.springframework.stereotype.Repository;

@Repository

public interface StudentRepository extends JpaRepository<Student, UUID> {

}

StudentController sẽ expose API thêm mới sinh viên có nội dung như sau:

package com.huongdanjava.systemdesign.outboxpattern.controller;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.huongdanjava.systemdesign.outboxpattern.repository.OutboxRepository;
import com.huongdanjava.systemdesign.outboxpattern.repository.StudentRepository;
import com.huongdanjava.systemdesign.outboxpattern.repository.entity.Outbox;
import com.huongdanjava.systemdesign.outboxpattern.repository.entity.Student;
import jakarta.transaction.Transactional;
import java.util.UUID;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class StudentController {

  private final StudentRepository studentRepository;
  private final OutboxRepository outboxRepository;

  private final ObjectMapper om = new ObjectMapper();

  public StudentController(StudentRepository studentRepository, OutboxRepository outboxRepository) {
    this.studentRepository = studentRepository;
    this.outboxRepository = outboxRepository;
  }

  @PostMapping("/students")
  @Transactional
  ResponseEntity<String> addNewStudent(@RequestParam String name) throws JsonProcessingException {
    Student student = new Student();
    student.setName(name);
    student.setId(UUID.randomUUID());
    Student s = studentRepository.save(student);

    Outbox outbox = new Outbox();
    outbox.setId(UUID.randomUUID());
    outbox.setAggregateId(s.getId());
    outbox.setEventType("StudentCreated");
    outbox.setType("Student");
    outbox.setPayload(om.writeValueAsString(student));
    outboxRepository.save(outbox);

    return ResponseEntity.ok("Added new student");
  }

}

package com.huongdanjava.systemdesign.outboxpattern.controller;

import com.fasterxml.jackson.core.JsonProcessingException;

import com.fasterxml.jackson.databind.ObjectMapper;

import com.huongdanjava.systemdesign.outboxpattern.repository.OutboxRepository;

import com.huongdanjava.systemdesign.outboxpattern.repository.StudentRepository;

import com.huongdanjava.systemdesign.outboxpattern.repository.entity.Outbox;

import com.huongdanjava.systemdesign.outboxpattern.repository.entity.Student;

import jakarta.transaction.Transactional;

import java.util.UUID;

import org.springframework.http.ResponseEntity;

import org.springframework.web.bind.annotation.PostMapping;

import org.springframework.web.bind.annotation.RequestParam;

import org.springframework.web.bind.annotation.RestController;

@RestController

public class StudentController {

private final StudentRepository studentRepository;

private final OutboxRepository outboxRepository;

private final ObjectMapper om = new ObjectMapper();

public StudentController(StudentRepository studentRepository, OutboxRepository outboxRepository) {

this.studentRepository = studentRepository;

this.outboxRepository = outboxRepository;

}

@PostMapping("/students")

@Transactional

ResponseEntity<String> addNewStudent(@RequestParam String name) throws JsonProcessingException {

Student student = new Student();

student.setName(name);

student.setId(UUID.randomUUID());

Student s = studentRepository.save(student);

Outbox outbox = new Outbox();

outbox.setId(UUID.randomUUID());

outbox.setAggregateId(s.getId());

outbox.setEventType("StudentCreated");

outbox.setType("Student");

outbox.setPayload(om.writeValueAsString(student));

outboxRepository.save(outbox);

return ResponseEntity.ok("Added new student");

}

Cùng với việc insert mới một record vào table student, chúng ta cũng sẽ insert 1 record vào outbox table. Table outbox này sẽ được monitor bởi Debezium để mỗi khi có record mới, Debezium sẽ pick up và gửi một message tới Apache Kafka.

Apache Kafka

Debezium

Tập tin cấu hình của Debezium cho ví dụ này của mình sẽ có nội dung như sau:

debezium.source.connector.class=io.debezium.connector.postgresql.PostgresConnector
debezium.source.database.hostname=localhost
debezium.source.database.port=5430
debezium.source.database.user=khanh
debezium.source.database.password=123456
debezium.source.database.dbname=outbox_pattern_example

debezium.source.topic.prefix=outbox_pattern

debezium.source.offset.storage.file.filename=data/offsets.dat
debezium.source.offset.flush.interval.ms=0

debezium.source.schema.history.internal=io.debezium.storage.file.history.FileSchemaHistory
debezium.source.schema.history.internal.file.filename=data/schema-history.dat

debezium.source.plugin.name=pgoutput

debezium.sink.type=kafka
debezium.sink.kafka.producer.bootstrap.servers=127.0.0.1:9092
debezium.sink.kafka.producer.key.serializer=org.apache.kafka.common.serialization.StringSerializer
debezium.sink.kafka.producer.value.serializer=org.apache.kafka.common.serialization.StringSerializer

debezium.source.connector.class=io.debezium.connector.postgresql.PostgresConnector

debezium.source.database.hostname=localhost

debezium.source.database.port=5430

debezium.source.database.user=khanh

debezium.source.database.password=123456

debezium.source.database.dbname=outbox_pattern_example

debezium.source.topic.prefix=outbox_pattern

debezium.source.offset.storage.file.filename=data/offsets.dat

debezium.source.offset.flush.interval.ms=0

debezium.source.schema.history.internal=io.debezium.storage.file.history.FileSchemaHistory

debezium.source.schema.history.internal.file.filename=data/schema-history.dat

debezium.source.plugin.name=pgoutput

debezium.sink.type=kafka

debezium.sink.kafka.producer.bootstrap.servers=127.0.0.1:9092

debezium.sink.kafka.producer.key.serializer=org.apache.kafka.common.serialization.StringSerializer

debezium.sink.kafka.producer.value.serializer=org.apache.kafka.common.serialization.StringSerializer

Với cấu hình trên, một topic có tên là outbox_pattern.public.outbox sẽ được tạo trong Apache Kafka đó các bạn! Tất cả những thay đổi của table outbox sẽ được capture và public lên topic này.

Các bạn xem thêm bài viết Change Data Capture với Debezium để hiểu rõ hơn về những thông tin cấu hình này nhé!

Service 2

Cho Service 2, mình cũng tạo một Spring Boot project sử dụng Spring for Apache Kafka để subscribe vào topic của Apache Kafka, cái topic mà Debezium sẽ public data change của Service 1. Sau đó thì sẽ in dữ liệu nhận được ra console.

Service 2 của mình như sau:

Tập tin application.yaml có nội dung như sau:

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: huongdanjava
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

spring:

kafka:

bootstrap-servers: localhost:9092

consumer:

group-id: huongdanjava

key-deserializer: org.apache.kafka.common.serialization.StringDeserializer

value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

Class listen topic outbox_pattern.public.outbox mình sẽ viết code đơn giản như sau:

package com.huongdanjava.systemdesign.outboxpattern;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class AppListener {

  @KafkaListener(topics = "outbox_pattern.public.outbox", groupId = "huongdanjava")
  public void listen(String message) {
    System.out.println("Received message: " + message);
  }

}

package com.huongdanjava.systemdesign.outboxpattern;

import org.springframework.kafka.annotation.KafkaListener;

import org.springframework.stereotype.Component;

@Component

public class AppListener {

@KafkaListener(topics = "outbox_pattern.public.outbox", groupId = "huongdanjava")

public void listen(String message) {

System.out.println("Received message: " + message);

}

Bây giờ, nếu chạy tất cả các ứng dụng của hệ thống, sau đó thì request tới Service 1 http://localhost:8081/students?name=Khanh với phương thức POST, các bạn sẽ thấy kết quả như sau: