message Entry { //Entry is the structure of a single subscription event. One event is equivalent to one binlog event in MySQL.
 Header header = 1;  //Event header
 Event event   = 2;  //Event body
}


message Header {
 int32       version        = 1;   //Entry protocol version
 SourceType  sourceType     = 2;   //The type of the source database, including MySQL, Oracle, and other types
 MessageType messageType    = 3;   //The type of the message, that is, the type of the Event, including BEGIN, COMMIT, DML, etc.
 uint32 timestamp           = 4;   //The timestamp of the Event in the original binlog
 int64 serverId = 5;   // The server ID of the source
 string fileName = 6;   // The file name of the source binlog
 uint64 position = 7;   // The offset of the event in the source binlog file
 string gtid = 8;   // The GTID of the current transaction
 string schemaName = 9;   // The schema affected by the change
 string tableName = 10;  // The table affected by the change
 uint64 seqId               = 11;  //Globally incremental sequence number
 uint64 eventIndex          = 12;  //If a large event is fragmented, each fragment is numbered from 0. This field is meaningless in the current version and reserved for future extension.
 bool   isLast              = 13;  //Whether the current event is the last fragment of the event. If so, the value is true. This field is meaningless in the current version and reserved for future extension.
 repeated KVPair properties = 15;
}


message Event {
 BeginEvent      beginEvent      = 1;  //The begin event in the binlog
 DMLEvent        dmlEvent        = 2;  // The dml event in the binlog
 CommitEvent     commitEvent     = 3;  // The commit event in the binlog
 DDLEvent        ddlEvent        = 4;  // The ddl event in the binlog
 RollbackEvent   rollbackEvent   = 5;  // The rollback event. This field is meaningless in the current version.
 HeartbeatEvent  heartbeatEvent  = 6;  // The heartbeat event periodically sent by the source database
 CheckpointEvent checkpointEvent = 7;  // The checkpoint event added by the subscription backend. One is automatically generated every 10 seconds for Kafka production and consumption offset management.
 repeated KVPair properties      = 15;
}

message Entries {
     repeated Entry items = 1; //entry list
}

message Envelope {
     int32 version                  = 1; // The protocol version, which determines how the data content is decoded.
     uint32 total                    = 2;
     uint32 index                    = 3;
     bytes  data                     = 4; // The current version is 1, indicating that the data in the data field is the result of PB serialization of Entries.
     repeated KVPair properties      = 15;
}

// Create a Flink-kafka consumer
FlinkKafkaConsumer<RecordMsgObject> consumer = 
                       new FlinkKafkaConsumer<>(topic, new DeserializeProtobufToRecordMsgObject(), props);


                       
// A custom deserializer that deserializes raw messages into RecordMsgObject objects.  
@Override
public RecordMsgObject deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
    RecordMsgObject obj = new RecordMsgObject();
    obj.topic = record.topic();
    obj.partition = record.partition();
    obj.offset = record.offset();
    obj.partitionSeq = getPartitionSeq(record);
    obj.key = new String(record.key());
    obj.headers = record.headers();
    // What is received here is the binary value of Envelope.
    obj.value = record.value();
    return obj;
}           

// Process the received messages by partition.
stream.keyBy(RecordMsgObject::getPartition)
      .process(new SubscribeMsgProcess(trans2sql)).setParallelism(1);

// Deserialize the envelope. In this demo, one envelope can store the data of an entire binlog event by default.
SubscribeProtobufData.Envelope envelope = SubscribeProtobufData.Envelope.parseFrom(record.value);
if (1 != envelope.getVersion()) {
    throw new IllegalStateException(String.format("unsupported version: %d", envelope.getVersion()));
}

// Deserialize to obtain Entries.
ByteString envelopeData = envelope.getData();
SubscribeProtobufData.Entries entries;
if (1 == envelope.getTotal()) {
    entries = SubscribeProtobufData.Entries.parseFrom(envelopeData.toByteArray());
} else {
    entries = SubscribeProtobufData.Entries.parseFrom(shardMsgMap.get(shardId).toByteArray());
    shardMsgMap.remove(shardId);
}

// Iterate through each Entry and print the sql statement based on the Entry type.
for (SubscribeProtobufData.Entry entry : entries.getItemsList()) {
    onEntry(record.partition, record.offset, ps, entry, trans2sql);
}