Debezium 和 Quarkus：通过 CDC 模式来避免双重写入

@Channel("movies")Emitter<Record<Long, String>> movieEmitter; private static ObjectMapper objectMapper = new ObjectMapper(); public Movie dualWriteInsert(Movie movie) throws JsonProcessingException {    // Inserts to DB    movie.persist();     // Send an event to movies topic    final String payloadJson = objectMapper.writeValueAsString(movie);    long id = movie.id;     movieEmitter.send(Record.of(id, payloadJson));

<dependency>   <groupId>io.debezium</groupId>   <artifactId>debezium-bom</artifactId>   <version>1.9.4.Final</version>   <type>pom</type>   <scope>import</scope></dependency>

<dependency>  <groupId>io.debezium</groupId>  <artifactId>debezium-ddl-parser</artifactId></dependency><dependency>  <groupId>io.debezium</groupId>  <artifactId>debezium-embedded</artifactId></dependency><!-- We connect to a MySQL database, so we need debezium MySQL connector --><dependency>  <groupId>io.debezium</groupId>  <artifactId>debezium-connector-mysql</artifactId></dependency>

<dependency>   <groupId>${quarkus.platform.group-id}</groupId>   <artifactId>quarkus-debezium-bom</artifactId>   <version>${quarkus.platform.version}</version>   <type>pom</type>   <scope>import</scope> </dependency>

<dependency>  <groupId>io.debezium</groupId>  <artifactId>debezium-quarkus-outbox</artifactId></dependency>

import javax.persistence.Entity; import io.quarkus.hibernate.orm.panache.PanacheEntity; @Entitypublic class Movie extends PanacheEntity {     // No worries Quarkus will change them   // to private and auto-generate getters/setters at compilation time   public String name;   public String director;   public String genre; }

import javax.inject.Inject;import javax.ws.rs.POST;import javax.ws.rs.Path; import org.jboss.logging.Logger; @Path("/movie")public class MovieResource {    // Service to insert the movie data into Movie and Outbox tables   @Inject   MovieService movieService;    // Injects the logger   @Inject   Logger logger;    // Http Post method to insert a movie   @POST   public Movie insert(Movie movie) {       logger.info("New Movie inserted " + movie.name);       System.out.println(":)");             return movieService.insertMovie(movie);   }}

import java.time.Instant; import com.fasterxml.jackson.databind.JsonNode;import com.fasterxml.jackson.databind.ObjectMapper;import com.fasterxml.jackson.databind.node.ObjectNode; import io.debezium.outbox.quarkus.ExportedEvent; public class MovieEvent implements ExportedEvent<String, JsonNode> {    private static ObjectMapper mapper = new ObjectMapper();    // Set the type enclosed inside the event   private static final String TYPE = "Movie";   // Set the event type   private static final String EVENT_TYPE = "MovieCreated";    private final long gameId;   private final JsonNode jsonNode;   private final Instant timestamp;    // Saves Game info in the class   public MovieEvent(Movie movie) {       this.gameId = movie.id;       this.timestamp = Instant.now();       // Saves game content in a string column in JSON format       this.jsonNode = convertToJson(movie);   }    @Override   public String getAggregateId() {       return String.valueOf(this.gameId);   }    @Override   public String getAggregateType() {       return TYPE;   }    @Override   public JsonNode getPayload() {       return jsonNode;   }    @Override   public Instant getTimestamp() {       return timestamp;   }    @Override   public String getType() {       return EVENT_TYPE;   }     private JsonNode convertToJson(Movie movie) {       ObjectNode asJson = mapper.createObjectNode()               .put("id", movie.id)               .put("name", movie.name)               .put("director", movie.director)               .put("genre", movie.genre);             return asJson;   } }

import javax.enterprise.context.ApplicationScoped;import javax.enterprise.event.Event;import javax.inject.Inject;import javax.transaction.Transactional; import io.debezium.outbox.quarkus.ExportedEvent; @ApplicationScopedpublic class MovieService {     // CDI event interface triggering Outbox entities   @Inject   Event<ExportedEvent<?, ?>> event;    // Transaction method   @Transactional   public Movie insertMovie(Movie movie) {        // Persists data       movie.persist();             // Persists outbox content       event.fire(new MovieEvent(movie));             return movie;   }}

import java.io.File;import java.io.IOException; import javax.enterprise.inject.Produces; import org.eclipse.microprofile.config.inject.ConfigProperty; import io.debezium.config.Configuration; public class DebeziumConfiguration {     // Debezium needs Database URL and credentials to login and   // monitor transaction logs   @ConfigProperty(name = "quarkus.datasource.jdbc.url")   String url;    @ConfigProperty(name = "quarkus.datasource.password")   String password;    @ConfigProperty(name = "quarkus.datasource.username")   String username;
   @Produces   public Configuration configureDebezium() throws IOException {        // Custom class to get database name or hostname of Database server       MySqlJdbcParser jdbcParser = MySqlJdbcParser.parse(url);             File fileOffset = File.createTempFile("offset", ".dat");       File fileDbHistory = File.createTempFile("dbhistory", ".dat");        return io.debezium.config.Configuration.create()           .with("name", "movies-mysql-connector")           // configures MySQL connector           .with("connector.class", "io.debezium.connector.mysql.MySqlConnector")           .with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")           .with("offset.storage.file.filename", fileOffset.getAbsolutePath())           .with("offset.flush.interval.ms", "60000")           // Configures database location           .with("database.hostname", jdbcParser.getHost())           .with("database.port", jdbcParser.getPort())           .with("database.user", "root")           .with("database.allowPublicKeyRetrieval", "true")           .with("database.password", password)           .with("database.dbname", jdbcParser.getDatabase())           .with("database.include.list", jdbcParser.getDatabase())           // Debezium only sends events for the modifications of OutboxEvent table and not all tables           .with("table.include.list", jdbcParser.getDatabase() + ".OutboxEvent")           .with("include.schema.changes", "false")           .with("database.server.id", "10181")           .with("database.server.name", "movies-mysql-db-server")           .with("database.history", "io.debezium.relational.history.FileDatabaseHistory")           .with("database.history.file.filename", fileDbHistory.getAbsolutePath())       .build();   } }

import java.io.IOException; import javax.enterprise.context.ApplicationScoped;import javax.enterprise.event.Observes; import org.apache.kafka.connect.data.Struct;import org.apache.kafka.connect.source.SourceRecord;import org.eclipse.microprofile.context.ManagedExecutor;import org.eclipse.microprofile.reactive.messaging.Channel;import org.eclipse.microprofile.reactive.messaging.Emitter; import com.fasterxml.jackson.core.JsonProcessingException;import com.fasterxml.jackson.databind.JsonNode;import com.fasterxml.jackson.databind.ObjectMapper; import io.debezium.config.Configuration;import io.debezium.embedded.Connect;import io.debezium.engine.DebeziumEngine;import io.debezium.engine.RecordChangeEvent;import io.debezium.engine.format.ChangeEventFormat;import io.quarkus.runtime.ShutdownEvent;import io.quarkus.runtime.StartupEvent;import io.smallrye.reactive.messaging.kafka.Record; import static io.debezium.data.Envelope.FieldName.*;import static io.debezium.data.Envelope.Operation;
@ApplicationScopedpublic class DebeziumListener {     private static ObjectMapper objectMapper = new ObjectMapper();    // Start the Debezium engine in a different thread   ManagedExecutor executor;    // Debezium configuration object   Configuration configuration;    private DebeziumEngine<RecordChangeEvent<SourceRecord>> engine;    public DebeziumListener(ManagedExecutor executor, Configuration configuration) {       this.executor = executor;       this.configuration = configuration;   }    // Interface to send events to movies Kafka topic   @Channel("movies")   Emitter<Record<Long, JsonNode>> movieEmitter;    void onStart(@Observes StartupEvent event) {        // Configures Debezium engine       this.engine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class))           .using(this.configuration.asProperties())           // For each event triggered by Debezium, the handleChangeEvnt method is called           .notifying(this::handleChangeEvent)           .build();        // Starts Debezium in different thread       this.executor.execute(this.engine);   }
   void handleChangeEvent(RecordChangeEvent<SourceRecord> sourceRecordRecordChangeEvent) {        // For each triggered event, we get the information       SourceRecord sourceRecord = sourceRecordRecordChangeEvent.record();       Struct sourceRecordChangeValue= (Struct) sourceRecord.value();        if (sourceRecordChangeValue != null) {           Operation operation = Operation.forCode((String) sourceRecordChangeValue.get(OPERATION));            // Only insert operations are processed           if(operation == Operation.CREATE) {                // Get insertation info               Struct struct = (Struct) sourceRecordChangeValue.get(AFTER);               String type = struct.getString("type");               String payload = struct.getString("payload");                if ("GameCreated".equals(type)) {                   try {                       final JsonNode payloadJson = objectMapper.readValue(payload, JsonNode.class);                       long id = payloadJson.get("id").asLong();                        // Populate content to Kafka topic                       movieEmitter.send(Record.of(id, payloadJson));                   } catch (JsonProcessingException e) {                       throw new IllegalArgumentException(e);                   }               }           }       }   }
   void onStop(@Observes ShutdownEvent event) throws IOException {       if (this.engine != null) {           this.engine.close();       }   } }

quarkus.hibernate-orm.log.sql=truequarkus.debezium-outbox.remove-after-insert=false

./mvnw clean quarkus:dev
…2022-07-07 11:36:22,942 INFO  [io.deb.con.mys.MySqlStreamingChangeEventSource] (debezium-mysqlconnector-movies-mysql-db-server-change-event-source-coordinator) Waiting for keepalive thread to start2022-07-07 11:36:22,948 INFO  [io.deb.con.mys.MySqlStreamingChangeEventSource] (debezium-mysqlconnector-movies-mysql-db-server-change-event-source-coordinator) Keepalive thread is running2022-07-07 11:37:43,889 INFO  [org.acm.MovieResource] (executor-thread-1) New Movie inserted string

docker ps
CONTAINER ID   IMAGE                          COMMAND                  CREATED          STATUS          PORTS                                             fa316bfae219   vectorized/redpanda:v21.11.3   "sh -c 'while [ ! -f…"   49 seconds ago   Up 45 seconds   8081-8082/tcp, 9644/tcp, 0.0.0.0:55002->9092/tcp
4c220f7ee066   mysql:8.0                      "docker-entrypoint.s…"   50 seconds ago   Up 46 seconds   33060/tcp, 0.0.0.0:60652->3306/tcp
e41cae02ff02   testcontainers/ryuk:0.3.3      "/app"                   53 seconds ago   Up 50 seconds   0.0.0.0:60650->8080/tcp

curl -X 'POST' \  'http://localhost:8080/movie' \  -H 'accept: application/json' \  -H 'Content-Type: application/json' \  -d '{  "name": "Minions: The Rise of Gru",  "director": "Kyle Balda",  "genre": "Animation"}'

:)Hibernate:    select        next_val as id_val    from        hibernate_sequence for update


Hibernate:    update        hibernate_sequence    set        next_val= ?    where        next_val=?
// Insert into Movie 
Hibernate:    insert    into        Movie        (director, genre, name, id)    values        (?, ?, ?, ?)
// Automatically OutboxEvent table receives an insert
Hibernate:    insert    into        OutboxEvent        (aggregatetype, aggregateid, type, timestamp, payload, tracingspancontext, id)    values        (?, ?, ?, ?, ?, ?, ?)

kcat -b localhost:55002 -C -t movies
{"id":1,"name":"Minions: The Rise of Gru","director":"Kyle Balda","genre":"Animation"}% Reached end of topic movies [0] at offset 1