message Entry { // An `Entry` is the structure of an individual subscription event. An event is similar to a binlog event in MySQL.
 Header header = 1;  // The event header
 Event event   = 2;  // The event body
}


message Header {
 int32       version        = 1;   // The protocol version of the `Entry`
 SourceType  sourceType     = 2;   // The source database type, such as MySQL and Oracle
 MessageType messageType    = 3;   // The message type, i.e., event type, such as BEGIN, COMMIT, and DML
 uint32 timestamp           = 4;   // The event timestamp in the source binlog
 int64  serverId            = 5;   // The `serverId` of the source database
 string fileName            = 6;   // The filename of the source binlog
 uint64 position            = 7;   // The event offset in the source binlog file
 string gtid                = 8;   // The GTID of the current transaction
 string schemaName          = 9;   // The modified schema
 string tableName           = 10;  // The modified table
 uint64 seqId               = 11;  // The globally incremental serial number
 uint64 eventIndex          = 12;  // If a large event is sharded, the shard number starts from 0. This parameter is meaningless on the current version and is reserved for future use.
 bool   isLast              = 13;  // Whether the current shard is the last shard of a sharded event; if so, the value is `true`. This parameter is meaningless on the current version and is reserved for future use.
 repeated KVPair properties = 15;
}


message Event {
 BeginEvent      beginEvent      = 1;  // The BIGIN 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 parameter is meaningless on the current version.
 HeartbeatEvent  heartbeatEvent  = 6;  // The heartbeat event regularly sent by the source database
 CheckpointEvent checkpointEvent = 7;  // The checkpoint event added to the subscription backend, which is generated automatically once every 10 seconds and is used 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; // Here, `version` is 1, indicating that the data is `Entries` serialized in the Protobuf format.
     repeated KVPair properties      = 15;
}

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


                       
// Customize a message deserializer to deserialize the original message to a `RecordMsgObject` object  
@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();
    // Here the binary value of `Envelope` will be received
    obj.value = record.value();
    return obj;
}           

//Put the received messages in different partitions
stream.keyBy(RecordMsgObject::getPartition)
      .process(new SubscribeMsgProcess(trans2sql)).setParallelism(1);

// Obtain an `Envlope` by deserialization. In this demo, only one `Envlope` can store the data of the whole 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 `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);
}

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