message Entry { //Entry は単一のサブスクリプションイベント構造です。1つのイベントは MySQL の1つの binlog event に相当します。
 Header header = 1;  //イベントヘッダー
 Event event   = 2;  //イベント本体
}


message Header {
 int32       version        = 1;   //Entry プロトコルバージョン
 SourceType  sourceType     = 2;   //ソースデータベースのタイプ情報。MySQL、Oracle などのタイプが含まれます。
 MessageType messageType    = 3;   //メッセージのタイプ、つまり Event のタイプです。BEGIN、COMMIT、DML などが含まれます。
 uint32 timestamp           = 4;   //Event の元の binlog におけるタイムスタンプ
 int64  serverId            = 5;   //ソースの serverId
 string fileName            = 6;   //ソース binlog のファイル名
 uint64 position            = 7;   //イベントのソース binlog ファイルにおけるオフセット
 string gtid                = 8;   //現在のトランザクションの gtid
 string schemaName          = 9;   //変更の影響を受けるスキーマ
 string tableName           = 10;  //変更の影響を受けるテーブル
 uint64 seqId               = 11;  //グローバル増分シーケンス番号
 uint64 eventIndex          = 12;  //大きな event をシャード化する場合、各シャードは0から番号が振られます。現在のバージョンでは意味がなく、今後の拡張用に予約されています。
 bool   isLast              = 13;  //現在の event が event シャードの最後のブロックであるかどうかを示します。そうである場合は true です。現在のバージョンでは意味がなく、今後の拡張用に予約されています。
 repeated KVPair properties = 15;
}


message Event {
 BeginEvent      beginEvent      = 1;  //binlog 内の begin イベント
 DMLEvent        dmlEvent        = 2;  //binlog 内の dml イベント
 CommitEvent     commitEvent     = 3;  //binlog 内の commit イベント
 DDLEvent        ddlEvent        = 4;  //binlog 内の ddl イベント
 RollbackEvent   rollbackEvent   = 5;  //rollback イベント。現在のバージョンでは意味がありません。
 HeartbeatEvent  heartbeatEvent  = 6;  //ソースデータベースが定期的に送信するハートビートイベント
 CheckpointEvent checkpointEvent = 7;  //購読バックエンドが追加する checkpoint イベント。10秒ごとに1つ自動生成され、Kafka の生産および消費場所の管理に使用されます。
 repeated KVPair properties      = 15;
}

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

message Envelope {
     int32  version                  = 1; //protocol version。data 内容のデコード方法を決定します。
     uint32 total                    = 2;
     uint32 index                    = 3;
     bytes  data                     = 4; //現在の version は1です。これは data 内のデータが Entries を PB シリアル化した結果であることを示します。
     repeated KVPair properties      = 15;
}

// Flink-kafka コンシューマーを作成します。
FlinkKafkaConsumer<RecordMsgObject> consumer = 
                       new FlinkKafkaConsumer<>(topic, new DeserializeProtobufToRecordMsgObject(), props);


                       
// カスタムデシリアライザ。元のメッセージを RecordMsgObject オブジェクトにデシリアライズします。  
@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();
    // ここで受信するのは Envelope のバイナリ値です。
    obj.value = record.value();
    return obj;
}           

//受信したメッセージを partition に従ってパーティション処理します。
stream.keyBy(RecordMsgObject::getPartition)
      .process(new SubscribeMsgProcess(trans2sql)).setParallelism(1);

// envelope をデシリアライズします。このデモでは、1つの envelope に binlog-event のデータ全体を格納できることがデフォルトです。
SubscribeProtobufData.Envelope envelope = SubscribeProtobufData.Envelope.parseFrom(record.value);
if (1 != envelope.getVersion()) {
    throw new IllegalStateException(String.format("unsupported version: %d", envelope.getVersion()));
}

// デシリアライズして 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);
}

// 各 Entry を走査し、Entry タイプに応じて sql を出力します。
for (SubscribeProtobufData.Entry entry : entries.getItemsList()) {
    onEntry(record.partition, record.offset, ps, entry, trans2sql);
}