Flink CDC作業(yè)失敗后能不能徹底退出，而不是重啟？

restart-strategy: fixed-delay
restart-strategy.fixed-delay.attempts: 2
restart-strategy.fixed-delay.delay: 10 s

MySQL/Hologres CDC源表不支持窗口函數(shù)，如何實(shí)現(xiàn)類似每分鐘聚合統(tǒng)計(jì)的需求?

SELECT 
    shop_id, 
    DATE_FORMAT(order_ts, 'yyyy-MM-dd HH:mm') AS window,
    COUNT(*) AS order_count, 
    SUM(price) AS amount 
FROM order_mysql_cdc 
GROUP BY shop_id, window

MySQL CDC表只能作為Source嗎？

是的。MySQL CDC源表可以讀取MySQL數(shù)據(jù)庫(kù)表中的全量和增量數(shù)據(jù)，只能作為Source使用。而MySQL表可以用作維表或者結(jié)果表。

MySQL CDC讀取全量數(shù)據(jù)后，不讀增量數(shù)據(jù)是為什么？

MySQL CDC使用table-name正則表達(dá)式不能解析逗號(hào)，怎么辦？

• 報(bào)錯(cuò)原因

  例如配置'table-name' = 't_process_wi_history_\d{1,2}'，報(bào)錯(cuò)如下。

• 報(bào)錯(cuò)原因

  Debezium使用逗號(hào)作為分隔符，不支持帶逗號(hào)的正則表達(dá)式，所以解析會(huì)報(bào)錯(cuò)。

• 解決方案

  建議您使用'table-name' = '(t_process_wi_history_\d{1}|t_process_wi_history_\d{2})'進(jìn)行配置。

作業(yè)重啟時(shí)，MySQL CDC源表會(huì)從作業(yè)停止時(shí)的位置消費(fèi)，還是從作業(yè)配置的位置重新消費(fèi)？

作業(yè)重啟時(shí)，您可以自由選擇啟動(dòng)策略。如果選擇全新啟動(dòng)，MySQL CDC源表會(huì)從配置的位置重新消費(fèi)。如果選擇從最新?tīng)顟B(tài)恢復(fù)，MySQL CDC源表會(huì)從作業(yè)停止時(shí)的位置開(kāi)始消費(fèi)。

比如，作業(yè)配置為從Binlog位點(diǎn){file=mysql-bin.01， position=40}啟動(dòng)作業(yè)，作業(yè)運(yùn)行一段時(shí)間后停止，此時(shí)消費(fèi)到Binlog位點(diǎn){file=mysql-bin.01， position=210}。如果選擇全新啟動(dòng)，MySQL CDC源表會(huì)重新從Binlog位點(diǎn){file=mysql-bin.01， position=40}消費(fèi)。如果選擇從最新?tīng)顟B(tài)恢復(fù)，則會(huì)從Binlog位點(diǎn){file=mysql-bin.01， position=210}開(kāi)始消費(fèi)。

MySQL CDC源表如何工作，會(huì)對(duì)數(shù)據(jù)庫(kù)造成什么影響？

在啟動(dòng)模式為initial（默認(rèn)值）時(shí)，MySQL CDC源表會(huì)先通過(guò)JDBC連接MySQL，使用SELECT語(yǔ)句讀取全量的數(shù)據(jù)，并記錄Binlog的位點(diǎn)。全量階段結(jié)束后，再?gòu)挠涗浀奈稽c(diǎn)開(kāi)始讀取Binlog中的增量數(shù)據(jù)。

show variables like '%max_connections%';

如何跳過(guò)Snapshot階段，只從變更數(shù)據(jù)開(kāi)始讀取？

可以通過(guò)WITH參數(shù)scan.startup.mode來(lái)控制，您可以指定從最早可用的Binlog位點(diǎn)消費(fèi)、從最新的Binlog位點(diǎn)消費(fèi)、指定時(shí)間戳消費(fèi)或指定具體的Binlog位點(diǎn)消費(fèi)，詳情可參見(jiàn)WITH參數(shù)：scan.startup.mode。

如何讀取一個(gè)分庫(kù)分表的MySQL數(shù)據(jù)庫(kù)？

如果MySQL是一個(gè)分庫(kù)分表的數(shù)據(jù)庫(kù)，分成了user_00、user_02和user_99等多個(gè)表，且所有表的schema一致。則可以通過(guò)table-name選項(xiàng)，指定一個(gè)正則表達(dá)式來(lái)匹配讀取多張表，例如設(shè)置table-name為user_.*，監(jiān)控所有user_前綴的表。database-name選項(xiàng)也支持該功能，但需要所有的表schema一致。

全表讀取階段效率慢、存在反壓，應(yīng)該如何解決？

如何判斷MySQL CDC作業(yè)是否已完成全量數(shù)據(jù)同步？

• 您可以通過(guò)監(jiān)控告警頁(yè)面currentEmitEventTimeLag指標(biāo)來(lái)判斷作業(yè)是否完成了全量數(shù)據(jù)同步。

  currentEmitEventTimeLag指標(biāo)記錄的是Source發(fā)送一條記錄到下游節(jié)點(diǎn)的時(shí)間點(diǎn)和該記錄在數(shù)據(jù)庫(kù)中生成時(shí)間點(diǎn)的差值，用于衡量數(shù)據(jù)從數(shù)據(jù)庫(kù)中產(chǎn)生到離開(kāi)Source節(jié)點(diǎn)的延遲。

  currentEmitEventTimeLag指標(biāo)取值含義如下：

  • 當(dāng)該指標(biāo)小于等于0時(shí)，則代表作業(yè)還在全量數(shù)據(jù)同步階段。

  • 當(dāng)該指標(biāo)大于0時(shí)，則代表作業(yè)完成了全量數(shù)據(jù)同步，進(jìn)入了Binlog讀取階段。

• 在MySQL CDC源表所在的TM日志中排查是否有BinlogSplitReader is created日志來(lái)判斷是否讀取完了全量數(shù)據(jù)，例如下圖所示。

  

多個(gè)CDC作業(yè)導(dǎo)致數(shù)據(jù)庫(kù)壓力過(guò)大怎么辦？

MySQL CDC源表需要連接數(shù)據(jù)庫(kù)讀取Binlog，當(dāng)源表數(shù)量逐漸增加，數(shù)據(jù)庫(kù)壓力也會(huì)逐漸增加。為了解決數(shù)據(jù)庫(kù)壓力過(guò)大的問(wèn)題，可以考慮通過(guò)將表同步到Kafka消息隊(duì)列中，再通過(guò)消費(fèi)Kafka中數(shù)據(jù)進(jìn)行解耦。詳情請(qǐng)參見(jiàn)MySQL整庫(kù)同步Kafka。

如果是通過(guò)CTAS方式同步數(shù)據(jù)導(dǎo)致的數(shù)據(jù)庫(kù)壓力過(guò)大，可以將多個(gè)CTAS作業(yè)合并為一個(gè)作業(yè)運(yùn)行。在配置相同的情況下，為每一個(gè)MySQL CDC源表配置相同Server ID，可以實(shí)現(xiàn)數(shù)據(jù)源的復(fù)用，從而減小數(shù)據(jù)庫(kù)的壓力。詳情請(qǐng)參見(jiàn)代碼示例四：多CTAS語(yǔ)句。

使用MySQL CDC，為什么數(shù)據(jù)量不大，但是flink讀取時(shí)候消耗了大量帶寬？

• 問(wèn)題詳情

  MySQL的源表數(shù)據(jù)更新量不大，但是Flink在讀取數(shù)據(jù)時(shí)消耗了大量的帶寬。

• 問(wèn)題原因

  在MySQL中，Binlog是整個(gè)實(shí)例級(jí)別的，因此它會(huì)記錄所有數(shù)據(jù)庫(kù)和所有表的變更。也就是說(shuō)，如果您的MySQL數(shù)據(jù)庫(kù)下有三張表，即使Flink作業(yè)只涉及其中一張表的變更，Binlog也會(huì)包含這三張表的所有變更記錄。

  使用Flink CDC時(shí)，雖然底層Binlog包含整個(gè)MySQL實(shí)例的所有變更數(shù)據(jù)，F(xiàn)link CDC可以通過(guò)配置只過(guò)濾并讀取特定表的變更記錄。這個(gè)過(guò)濾過(guò)程是在Debezium或Flink CDC連接器層面完成的，而不是在MySQL層面完成的。

• 解決方案

  雖然無(wú)法解決Binlog的數(shù)據(jù)變更存儲(chǔ)機(jī)制，但是可以通過(guò)Source復(fù)用避免更多的帶寬使用，詳情請(qǐng)參考開(kāi)啟CDC Source復(fù)用。

使用MySQL CDC，增量階段讀取出來(lái)的timestamp字段時(shí)區(qū)相差8小時(shí)，怎么回事呢？

• 在解析Binlog數(shù)據(jù)中的timestamp字段時(shí)，CDC作業(yè)里配置server-time-zone參數(shù)，如果這個(gè)參數(shù)沒(méi)有和您的MySQL服務(wù)器時(shí)區(qū)一致，就會(huì)出現(xiàn)這個(gè)問(wèn)題。

• 在DataStream中使用了自定義序列化器，例如MyDeserializer implements DebeziumDeserializationSchema。當(dāng)自定義序列化器解析timestamp類型的數(shù)據(jù)時(shí)，出現(xiàn)該問(wèn)題。可以參考RowDataDebeziumDeserializeSchema中對(duì)timestamp類型的解析，在serverTimeZone處給定時(shí)區(qū)信息。

    private TimestampData convertToTimestamp(Object dbzObj, Schema schema) {
            if (dbzObj instanceof Long) {
                switch (schema.name()) {
                    case Timestamp.SCHEMA_NAME:
                       return TimestampData.fromEpochMillis((Long) dbzObj);
                    case MicroTimestamp.SCHEMA_NAME:
                       long micro = (long) dbzObj;
                       return TimestampData.fromEpochMillis(micro / 1000, (int) (micro % 1000 * 1000));
                    case NanoTimestamp.SCHEMA_NAME:
                       long nano = (long) dbzObj;
                       return TimestampData.fromEpochMillis(nano / 1000_000, (int) (nano % 1000_000));
                }
           }
           LocalDateTime localDateTime = TemporalConversions.toLocalDateTime(dbzObj, serverTimeZone);
           return TimestampData.fromLocalDateTime(localDateTime);
      }

MySQL CDC支持監(jiān)聽(tīng)從庫(kù)嗎？從庫(kù)需要如何配置？

是的，為了支持MySQL CDC監(jiān)聽(tīng)從庫(kù)，需要對(duì)從庫(kù)進(jìn)行如下配置。配置完成后，從庫(kù)在接收主庫(kù)同步過(guò)來(lái)的數(shù)據(jù)時(shí)，會(huì)將這些數(shù)據(jù)寫(xiě)入從庫(kù)自身的Binlog文件中。

log-slave-updates = 1

如果主庫(kù)啟用了GTID模式，那么從庫(kù)也必須啟用GTID模式。對(duì)主從庫(kù)進(jìn)行如下配置。

gtid_mode = on
enforce_gtid_consistency = on

怎么獲取數(shù)據(jù)庫(kù)中的DDL事件？

使用社區(qū)版本的CDC連接器時(shí)，您可以通過(guò)DataStream API使用MySqlSource，配置includeSchemaChanges(true)參數(shù)來(lái)獲取DDL事件。獲取到DDL事件后，再編寫(xiě)相應(yīng)的代碼進(jìn)行后續(xù)處理。代碼示例如下。

MySqlSource<xxx> mySqlSource =
 MySqlSource.<xxx>builder()
 .hostname(...)
 .port(...)
 .databaseList("<databaseName>")
 .tableList("<databaseName>.<tableName>")
 .username(...)
 .password(...)
 .serverId(...)
 .deserializer(...)
 .includeSchemaChanges(true) // 配置獲取DDL事件的參數(shù)
 .build();
 ... // 其他處理邏輯 

Flink CDC支持MySQL整庫(kù)同步嗎？怎么做？

支持的，實(shí)時(shí)計(jì)算Flink版提供了CTAS或CDAS語(yǔ)法來(lái)支持整庫(kù)同步，詳情請(qǐng)參見(jiàn)CREATE TABLE AS（CTAS）語(yǔ)句或CREATE DATABASE AS（CDAS）語(yǔ)句。

同一個(gè)實(shí)例下，某個(gè)庫(kù)的表無(wú)法同步增量數(shù)據(jù)，其他庫(kù)都可以，為什么？

因?yàn)镸ySQL服務(wù)器可以配置Binlog過(guò)濾器，忽略了某些庫(kù)的Binlog。您可以通過(guò)show master status命令查看Binlog_Ignore_DB和Binlog_Do_DB。查看結(jié)果示例如下。

mysql> show master status;
+------------------+----------+--------------+------------------+----------------------+
| File             | Position | Binlog_Do_DB | Binlog_Ignore_DB |  Executed_Gtid_Set   |
+------------------+----------+--------------+------------------+----------------------+
| mysql-bin.000006 |     4594 |              |                  | xxx:1-15             |
+------------------+----------+--------------+------------------+----------------------+

使用DataStream API構(gòu)建MySQL CDC Source時(shí)如何配置tableList選項(xiàng)？

tableList要求表名包含數(shù)據(jù)庫(kù)名，而不是DataStream API中的表名。對(duì)于MySQL CDC Source，可以配置tableList("yourDatabaseName.yourTableName")參數(shù)。?

MongoDB CDC全量讀取階段，作業(yè)失敗后，可以從checkpoint繼續(xù)讀取嗎？

在WITH參數(shù)中配置'scan.incremental.snapshot.enabled'= 'true'參數(shù)，可以從checkpoint恢復(fù)讀取數(shù)據(jù)。

MongoDB CDC支持全量+增量讀和只讀增量嗎？

支持，默認(rèn)為全量+增量讀取；在WITH參數(shù)中配置'scan.startup.mode' = 'latest-offset'參數(shù)將讀取數(shù)據(jù)模式變?yōu)橹蛔x增量。

MongoDB CDC支持訂閱多個(gè)集合嗎？

僅支持訂閱整個(gè)庫(kù)的集合，不支持篩選部分集合功能。例如，在WITH參數(shù)中設(shè)置'database' = 'mgdb'，并且'collection' = ''，則表示會(huì)訂閱mgdb數(shù)據(jù)庫(kù)下的所有集合。

MongoDB CDC支持設(shè)置多并發(fā)度讀取嗎？

如果啟用了scan.incremental.snapshot.enabled配置，則在初始快照階段將支持并發(fā)讀取。

MongoDB CDC支持的MongoDB版本是哪些？

MongoDB CDC基于Change Stream特性實(shí)現(xiàn)，Change Stream是MongoDB 3.6新增的特性。MongoDB CDC理論上支持3.6及更高的版本，建議使用版本>=4.0。如果在低于3.6的版本上運(yùn)行，可能會(huì)出現(xiàn)錯(cuò)誤Unrecognized pipeline stage name: '$changeStream'。

MongoDB CDC支持的MongoDB運(yùn)行模式是什么樣的？

Change Stream需要MongoDB在副本集或分片集群架構(gòu)下運(yùn)行，在本地測(cè)試時(shí)，可以使用單節(jié)點(diǎn)副本集架構(gòu)，通過(guò)執(zhí)行rs.initiate()命令來(lái)初始化。但需要注意，在單節(jié)點(diǎn)架構(gòu)下執(zhí)行CDC可能會(huì)出現(xiàn)錯(cuò)誤The $changestage is only supported on replica sets。

MongoDB CDC是否支持Debezium相關(guān)的參數(shù)？

不支持的，因?yàn)镸ongoDB CDC連接器是在Flink CDC項(xiàng)目中獨(dú)立開(kāi)發(fā)，并不依賴Debezium項(xiàng)目。

其他組件可以成功使用相同的用戶名和密碼進(jìn)行連接，為何MongoDB CDC會(huì)報(bào)錯(cuò)并提示用戶名密碼錯(cuò)誤？

因?yàn)樵撚脩魬{證是在所需連接的數(shù)據(jù)庫(kù)下創(chuàng)建的。如果需要正常訪問(wèn)，需要在WITH參數(shù)里添加 'connection.options' = 'authSource=用戶所在的DB'。

MongoDB CDC支持從Checkpoint恢復(fù)嗎？原理是怎么樣的呢？

支持，checkpoint會(huì)記錄Change Stream的Resume Token，可以通過(guò)Resume Token重新恢復(fù)Change Stream。其中Resume Token對(duì)應(yīng)oplog.rs（MongoDB變更日志集合）的位置，oplog.rs是一個(gè)固定容量的集合。

當(dāng)Resume Token對(duì)應(yīng)的記錄在oplog.rs中不存在時(shí)，可能會(huì)出現(xiàn)無(wú)效Resume Token的異常，您可以設(shè)置合適oplog.rs的集合大小，避免oplog.rs保留時(shí)間過(guò)短，詳情請(qǐng)參見(jiàn)更改Oplog的大小。

另外，新達(dá)到的變更記錄和心跳記錄可以刷新Resume Token。

MongoDB CDC支持輸出-U（update_before，更新前鏡像值）消息嗎？

• 在MongoDB  6.0及以上版本中，若數(shù)據(jù)庫(kù)開(kāi)啟了前像或后像功能，您可以在SQL作業(yè)中配置參數(shù)'scan.full-changelog' = 'true'，使得數(shù)據(jù)源能夠輸出-U消息，從而省去ChangelogNormalize。

• 在MongoDB 6.0以下版本中，由于MongoDB原始的oplog.rs包含INSERT、UPDATE、REPLACE和DELETE這四種操作類型，沒(méi)有保留更新前的信息，因此無(wú)法直接輸出-U類型消息。在Flink中，只能實(shí)現(xiàn)基于update的語(yǔ)義。使用MongoDBTableSource時(shí)，F(xiàn)link planner會(huì)自動(dòng)進(jìn)行ChangelogNormalize優(yōu)化，補(bǔ)齊缺失的-U消息，輸出完整的+I、 -U、+U、-D四種消息，ChangelogNormalize優(yōu)化的代價(jià)是該節(jié)點(diǎn)會(huì)保存之前所有key的狀態(tài)。如果是DataStream作業(yè)直接使用MongoDBSource，沒(méi)有Flink planner的優(yōu)化，將不會(huì)自動(dòng)進(jìn)行ChangelogNormalize優(yōu)化，所以不能直接獲取-U消息。若想要獲取更新前鏡像值，您需要自己管理狀態(tài)。如果不希望自己管理狀態(tài)，您可以將MongoDBTableSource中原始的oplog.rs轉(zhuǎn)換為ChangelogStream或者RetractStream，并借助Flink planner的優(yōu)化能力補(bǔ)齊更新前的鏡像值，示例如下。

   tEnv.executeSql("CREATE TABLE orders ( ... ) WITH ( 'connector'='mongodb-cdc',... )");
  
   Table table = tEnv.from("orders")
   .select($("*"));
  
   tEnv.toChangelogStream(table)
   .print()
   .setParallelism(1);
  
   env.execute();

如何配置參數(shù)以過(guò)濾作業(yè)中的非法日期臟數(shù)據(jù)？

可以在Postgres CDC的WITH參數(shù)中配置如下參數(shù)來(lái)過(guò)濾臟數(shù)據(jù)。

• 配置'debezium.event.deserialization.failure.handling.mode'='warn'參數(shù)，跳過(guò)臟數(shù)據(jù)，將臟數(shù)據(jù)打印到WARN日志里。

• 配置'debezium.event.deserialization.failure.handling.mode'='ignore'參數(shù)，跳過(guò)臟數(shù)據(jù)，不打印臟數(shù)據(jù)到日志。

Postgres CDC提示未傳輸TOAST數(shù)據(jù)，是什么原因？

請(qǐng)確保副本身份是完整的。TOAST數(shù)據(jù)相對(duì)較大，為了節(jié)省WAL的大小，如果TOAST數(shù)據(jù)沒(méi)有發(fā)生變化，您配置了'debezium.schema.refresh.mode'='columns_diff_exclude_unchanged_toast'參數(shù)，wal2json插件不會(huì)將TOAST數(shù)據(jù)帶到更新后的數(shù)據(jù)中。

發(fā)現(xiàn)PG服務(wù)器磁盤(pán)使用率高，WAL日志不釋放是什么原因？

Postgres CDC只會(huì)在checkpoint完成的時(shí)候更新Postgres slot中的LSN。如果發(fā)現(xiàn)磁盤(pán)使用率高的情況下，請(qǐng)先確認(rèn)Postgres的checkpoint是否開(kāi)啟并查看數(shù)據(jù)庫(kù)是否有其他未使用或同步延遲的slot。

Postgres CDC同步Postgres中DECIMAL類型數(shù)據(jù)精度超過(guò)最大精度時(shí)，會(huì)返回什么結(jié)果？

Postgres CDC中如果收到DECIMAL類型數(shù)據(jù)的精度大于在Postgres CDC中聲明的類型的精度時(shí)，會(huì)將數(shù)據(jù)處理成NULL。此時(shí)，您可以配置'debezium.decimal.handling.mode' = 'string'參數(shù)，將讀取的數(shù)據(jù)用字符串類型來(lái)進(jìn)行處理。

在DataStream API中構(gòu)建Postgres CDC Source時(shí)如何配置tableList選項(xiàng)？

tableList要求表名使用模式名，而不是DataStream API中的表名。對(duì)于Postgres CDC Source，tableList選項(xiàng)值為my_schema.my_table。

為什么無(wú)法下載flink-sql-connector-mysql-cdc-2.2-SNAPSHOT.jar，Maven倉(cāng)庫(kù)為什么沒(méi)有xxx-SNAPSHOT依賴？

和主流的Maven項(xiàng)目版本管理相同，xxx-SNAPSHOT版本對(duì)應(yīng)開(kāi)發(fā)分支的代碼。如果需要使用這個(gè)版本，您需要下載源碼并編譯對(duì)應(yīng)的JAR。您可以使用穩(wěn)定版本，例如flink-sql-connector-mysql-cdc-2.1.0.jar，可以直接從Maven中心倉(cāng)庫(kù)中獲取并使用。

使用flink-sql-connector-xxx.jar和使用flink-connector-xxx.jar有什么區(qū)別?

Flink CDC各個(gè)連接器的包命名規(guī)則和Flink其他連接器的包命名規(guī)則是保持一致的。

• flink-sql-connector-xx是FAT JAR，除了連接器的代碼外，還將連接器依賴的所有第三方包shade后打入FAT JAR，提供給SQL作業(yè)使用，您只需要在lib目錄下添加該FAT JAR即可。

• flink-connector-xx只包含該連接器的代碼，不包含其所需的依賴，供DataStream作業(yè)使用，您需要自己管理所需的第三方包依賴，有沖突的依賴需要進(jìn)行exclude和shade處理。

為什么Maven倉(cāng)庫(kù)里找不到2.x版本？

Flink CDC項(xiàng)目從2.0.0版本將group id從com.alibaba.ververica改成com.ververica，所以在Maven倉(cāng)庫(kù)找2.x版本的包時(shí)，路徑是/com/ververica。

DataStream API使用JsonDebeziumDeserializationSchema反序列化器時(shí)，數(shù)值類型顯示是一堆字符串，怎么辦？

因?yàn)镈ebezium在解析數(shù)值類型時(shí)有不同的轉(zhuǎn)換方式，詳情請(qǐng)參見(jiàn)Debezium connector for MySQL。在Flink CDC配置的轉(zhuǎn)換代碼如下。

Properties properties = new Properties();
....
properties.setProperty("bigint.unsigned.handling.mode","long");
properties.setProperty("decimal.handling.mode","double");

MySqlSource.<String>builder()
 .hostname(config.getHostname())
 ....
 .debeziumProperties(properties);

報(bào)錯(cuò)：Replication slot "xxxx" is active

• 報(bào)錯(cuò)詳情

  當(dāng)您的Postgres CDC作業(yè)結(jié)束后，可能會(huì)遇到slot未被正確釋放的情況。

• 解決方案

  您可以采用以下兩種方法釋放slot：

  • 在PostgreSQL中執(zhí)行以下命令手動(dòng)釋放slot。

    select pg_drop_replication_slot('rep_slot');

    如果遇到錯(cuò)誤“ERROR: replication slot "rep_slot" is active for PID 162564”，則表示slot正在被進(jìn)程（PID）占用。您需要先終止該進(jìn)程，再釋放slot。釋放命令如下所示。

    select pg_terminate_backend(162564);
    select pg_drop_replication_slot('rep_slot');

  • 自動(dòng)清理slot。在作業(yè)的Postgres Source配置中添加'debezium.slot.drop.on.stop' = 'true'參數(shù)，確保當(dāng)CDC作業(yè)停止時(shí)，slot會(huì)被自動(dòng)清理。

    警告

    如果開(kāi)啟自動(dòng)清理slot,  會(huì)導(dǎo)致Wal Log被回收，當(dāng)作業(yè)再次啟動(dòng)時(shí)，會(huì)導(dǎo)致數(shù)據(jù)丟失，無(wú)法保證At-Least Once語(yǔ)義。

報(bào)錯(cuò)：Lock wait timeout exceeded; try restarting transaction

• 報(bào)錯(cuò)詳情

  org.apache.kafka.connect.errors.ConnectException: Lock wait timeout exceeded; try restarting transaction Error code: 1205; SQLSTATE: 40001.
      at io.debezium.connector.mysql.AbstractReader.wrap(AbstractReader.java:241)
      at io.debezium.connector.mysql.AbstractReader.failed(AbstractReader.java:218)
      at io.debezium.connector.mysql.SnapshotReader.execute(SnapshotReader.java:857)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1147)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:622)
      at java.lang.Thread.run(Thread.java:834)
  Caused by: com.mysql.cj.jdbc.exceptions.MySQLTransactionRollbackException: Lock wait timeout exceeded; try restarting transaction
      at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:123)
      at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:97)
      at com.mysql.cj.jdbc.exceptions.SQLExceptionsMapping.translateException(SQLExceptionsMapping.java:122)
      at com.mysql.cj.jdbc.StatementImpl.executeQuery(StatementImpl.java:1200)
      at io.debezium.jdbc.JdbcConnection.query(JdbcConnection.java:554)
      at io.debezium.jdbc.JdbcConnection.query(JdbcConnection.java:497)
      at io.debezium.connector.mysql.SnapshotReader.readTableSchema(SnapshotReader.java:888)
      at io.debezium.connector.mysql.SnapshotReader.execute(SnapshotReader.java:550)
      ... 3 more

• 報(bào)錯(cuò)原因

  當(dāng)MySQL CDC源表不開(kāi)啟增量快照讀取時(shí)，申請(qǐng)鎖時(shí)可能發(fā)生的超時(shí)錯(cuò)誤。

• 解決方案

  升級(jí)到VVR 4.0.8及以上版本即可，新版本默認(rèn)使用無(wú)鎖算法，不需要申請(qǐng)鎖。

報(bào)錯(cuò)：Cause by：java.lang.ArrayIndexOutOfBoundsException

• 報(bào)錯(cuò)詳情

• 報(bào)錯(cuò)原因

  VVR 4.0.12以下版本依賴的Binlog讀取工具存在問(wèn)題，導(dǎo)致報(bào)錯(cuò)。

• 解決方案

  VVR 4.0.12及以上版本已修復(fù)此問(wèn)題，建議升級(jí)版本解決。

報(bào)錯(cuò)：Caused by: io.debezium.DebeziumException: Received DML 'xxx' for processing, binlog probably contains events generated with statement or mixed based replication format

• 報(bào)錯(cuò)詳情

  Caused by: io.debezium.DebeziumException: Received DML 'insert into gd_chat_fetch_log (
  
  id,
  c_cursor,
  d_timestamp,
  msg_cnt,
  state,
  ext1,
  ext2,
  cost_time
  
  ) values (
  null,
  null,
  '2022-03-23 16:51:00.616',
  0,
  1,
  null,
  null,
  0
  )' for processing, binlog probably contains events generated with statement or mixed based replication format

• 報(bào)錯(cuò)原因

  Binlog格式是Mixed導(dǎo)致報(bào)錯(cuò)。MySQL CDC源表要求Binlog的格式為ROW。

• 解決方案

  1. 使用show global variables like "binlog_format"命令，查看Binlog的格式。

     說(shuō)明

     該show variables like "binlog_format"命令只能查看當(dāng)前的Binlog格式，具有局限性。

  2. 如果Binlog格式不是ROW，您需要在MySQL Server端將Binlog格式修改為ROW。詳情請(qǐng)參見(jiàn)Setting The Binary Log Format。

  3. 重啟作業(yè)。

報(bào)錯(cuò)：Encountered change event for table xxx.xxx whose schema isn't known to this connector

• 報(bào)錯(cuò)詳情

  202x-xx-xx xx:xx:xx,xxx ERROR io.debezium.connector.mysql.BinlogReader                     [] - Encountered change event 'Event{header=EventHeaderV4{timestamp=xxx, eventType=TABLE_MAP, serverId=xxx, headerLength=xxx, dataLength=xxx, nextPosition=xxx, flags=xxx}, data=TableMapEventData{tableId=xxx, database='xxx', table='xxx', columnTypes=xxx, xxx..., columnMetadata=xxx,xxx..., columnNullability={xxx,xxx...}, eventMetadata=null}}' at offset {ts_sec=xxx, file=mysql-bin.xxx, pos=xxx, gtids=xxx, server_id=xxx, event=xxx} for table xxx.xxx whose schema isn't known to this connector. One possible cause is an incomplete database history topic. Take a new snapshot in this case.
  Use the mysqlbinlog tool to view the problematic event: mysqlbinlog --start-position=30946 --stop-position=31028 --verbose mysql-bin.004419
  202x-xx-xx xx:xx:xx,xxx ERROR io.debezium.connector.mysql.BinlogReader                     [] - Error during binlog processing. Last offset stored = null, binlog reader near position = mysql-bin.xxx/xxx
  202x-xx-xx xx:xx:xx,xxx ERROR io.debezium.connector.mysql.BinlogReader                     [] - Failed due to error: Error processing binlog event
  org.apache.kafka.connect.errors.ConnectException: Encountered change event for table statistic.apk_info whose schema isn't known to this connector
      at io.debezium.connector.mysql.AbstractReader.wrap(AbstractReader.java:241) ~[ververica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
      at io.debezium.connector.mysql.AbstractReader.failed(AbstractReader.java:218) ~[ververica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
      at io.debezium.connector.mysql.BinlogReader.handleEvent(BinlogReader.java:607) ~[ververica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
      at com.github.shyiko.mysql.binlog.BinaryLogClient.notifyEventListeners(BinaryLogClient.java:1104) [ververica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
      at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:955) [ververica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
      at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:595) [ververica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
      at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:839) [ververica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
      at java.lang.Thread.run(Thread.java:834) [?:1.8.0_102]
  Caused by: org.apache.kafka.connect.errors.ConnectException: Encountered change event for table xxx.xxx whose schema isn't known to this connector
      at io.debezium.connector.mysql.BinlogReader.informAboutUnknownTableIfRequired(BinlogReader.java:875) ~[ververica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
      at io.debezium.connector.mysql.BinlogReader.handleUpdateTableMetadata(BinlogReader.java:849) ~[ververica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
      at io.debezium.connector.mysql.BinlogReader.handleEvent(BinlogReader.java:590) ~[ververica-connector-mysql-1.12-vvr-3.0.0-SNAPSHOT-jar-with-dependencies.jar:1.12-vvr-3.0.0-SNAPSHOT]
      ... 5 more

• 報(bào)錯(cuò)原因

  • 當(dāng)作業(yè)中使用的某些數(shù)據(jù)庫(kù)，您沒(méi)有其對(duì)應(yīng)的某些權(quán)限時(shí)，可能出現(xiàn)該錯(cuò)誤。

  • 當(dāng)使用了'debezium.snapshot.mode'='never'時(shí)，可能出現(xiàn)該錯(cuò)誤。因?yàn)檫@會(huì)導(dǎo)致從Binlog開(kāi)頭讀取數(shù)據(jù)，但是Binlog開(kāi)頭的變更事件對(duì)應(yīng)的Table Schema和當(dāng)前表的Schema可能不匹配，所以會(huì)報(bào)該錯(cuò)誤。

  • 遇到Debezium解析不了的變更，例如Debezium無(wú)法解析`DEFAULT (now())`，可能出現(xiàn)該錯(cuò)誤。

• 解決方案

  • 先確認(rèn)您使用的數(shù)據(jù)庫(kù)用戶是否有對(duì)應(yīng)作業(yè)中全部數(shù)據(jù)庫(kù)的相應(yīng)權(quán)限，詳細(xì)授權(quán)操作請(qǐng)參見(jiàn)配置MySQL。

  • 不建議使用'debezium.snapshot.mode'='never'，可以通過(guò)'debezium.inconsistent.schema.handling.mode' = 'warn'參數(shù)避免報(bào)錯(cuò)。

  • 繼續(xù)觀察日志，查詢io.debezium.connector.mysql.MySqlSchema WARN的log，會(huì)打印出具體無(wú)法解析的變更詳情，例如Debezium解析不了`DEFAULT (now())`。

報(bào)錯(cuò)：org.apache.kafka.connect.errors.DataException: xxx is not a valid field name

• 報(bào)錯(cuò)詳情

  org.apache.kafka.connect.errors.DataException: xxx is not a valid field name
      at org.apache.kafka.connect.data.Struct.lookupField(Struct.java:254)
      at org.apache.kafka.connect.data.Struct.get(Struct.java:74)
      at com.alibaba.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema.lambda?createRowConverter$508c5858$1(RowDataDebeziumDeserializeSchema.java:369)
      at com.alibaba.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema.lambda?wrapIntoNullableConverter$7b91dc26$1(RowDataDebeziumDeserializeSchema.java:394)
      at com.alibaba.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema.extractAfterRow(RowDataDebeziumDeserializeSchema.java:127) 
      at com.alibaba.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema.deserialize(RowDataDebeziumDeserializeSchema.java:102)
      at com.alibaba.ververica.cdc.debezium.internal.DebeziumChangeConsumer.handleBatch(DebeziumChangeConsumer.java:124)
      at io.debezium.embedded.ConvertingEngineBuilder.lambda?notifying$2(ConvertingEngineBuilder.java:82) 
      at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:812) 
      at io.debezium.embedded.ConvertingEngineBuilder$2.run(ConvertingEngineBuilder.java:171)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1147)
      at java.util.concurrent.ThreadPoolExecutorsWorker.run ThreadPoolExecutor.java:622) 
      at java.lang.Thread.run(Thread.java: 834)

• 報(bào)錯(cuò)原因

  分庫(kù)分表中的某張物理表缺少了您在MySQL CDC表中定義的字段，導(dǎo)致Schema不一致。

  例如您使用正則表達(dá)式mydb.users_\d{3}去監(jiān)控mydb數(shù)據(jù)庫(kù)下users_001，users_002，……，users_999這些表，您在MySQL CDC表的DDL中聲明了user_name字段，但如果users_002表中無(wú)user_name字段，在解析到users_002表的Binlog時(shí)就會(huì)出現(xiàn)該異常。

• 解決方案

  針對(duì)分庫(kù)分表場(chǎng)景，分庫(kù)分表里的每個(gè)表中必須包含MySQL CDC DDL中聲明的字段。

  此外，您也可以將作業(yè)升級(jí)到VVR 6.0.2及以上版本，VVR 6.0.2及以上版本的MySQL CDC會(huì)自動(dòng)使用分庫(kù)分表中最寬的Schema解決該問(wèn)題。

報(bào)錯(cuò)：Caused by: java.sql.SQLSyntaxErrorException: Unknown storage engine 'BLACKHOLE'

• 報(bào)錯(cuò)詳情

• 報(bào)錯(cuò)原因

  在解析MySQL 5.6的DDL時(shí)，存在不支持的語(yǔ)法導(dǎo)致報(bào)錯(cuò)。

• 解決方案

  • 您可以在MySQL CDC表的WITH參數(shù)中加上'debezium.database.history.store.only.monitored.tables.ddl'='true'和'debezium.database.exclude.list'='mysql'兩個(gè)參數(shù)來(lái)避免報(bào)錯(cuò)。

  • 您也可以將作業(yè)升級(jí)到VVR 6.0.2及以上版本，VVR 6.0.2及以上版本的MySQL CDC對(duì)DDL解析提供了更好的支持。

報(bào)錯(cuò)：Caused by: org.apache.kafka.connect.errors.ConnectException: The connector is trying to read binlog starting at GTIDs ..., but this is no longer available on the server. Reconfigure the connector to use a snapshot when needed.

• 報(bào)錯(cuò)詳情

  org.apache.kafka.connect.errors.ConnectException: The connector is trying to read binlog starting at GTIDs xxx and binlog file 'binlog.000064', pos=89887992, skipping 4 events plus 1 rows, but this is no longer available on the server. Reconfigure the connector to use a snapshot when needed
      at io.debezium.connector.mysql.MySqlConnectorTask.start(MySqlConnectorTask.java:133)
      at io.debezium.connector.common. BaseSourceTask.start (BaseSourceTask.java:106) 
      at io.debezium.embedded.EmbeddedEngine.run (EmbeddedEngine.java:758) 
      at io.debezium.embedded.ConvertingEngineBuilder$2.run(ConvertingEngineBuilder.java:171)
      at java.util.concurrent.ThreadPoolExecutor. runWorker(ThreadPoolExecutor.java:1147)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:622)
      at java.lang.Thread.run(Thread.java:834)

• 報(bào)錯(cuò)原因和解決方案

  報(bào)錯(cuò)原因	解決方案
  作業(yè)正在讀取的Binlog文件在MySQL服務(wù)器已經(jīng)被清理掉時(shí)，會(huì)出現(xiàn)該報(bào)錯(cuò)。這種情況一般由于MySQL服務(wù)器上Binlog的保留時(shí)間太短。	可以將Binlog的保留時(shí)間調(diào)大，比如7天。具體命令如下。 show variables like 'expire_logs_days'; set global expire_logs_days=7;
  MySQL CDC作業(yè)消費(fèi)Binlog太慢，例如下游的聚合算子或者Sink算子長(zhǎng)時(shí)間出現(xiàn)反壓，反壓傳遞到source，導(dǎo)致source無(wú)法消費(fèi)數(shù)據(jù)。	需要對(duì)作業(yè)資源調(diào)優(yōu)，讓source恢復(fù)正常消費(fèi)即可。
  阿里云RDS MySQL的日志保留策略一般有兩個(gè)條件：最長(zhǎng)18個(gè)小時(shí)，最大占用30%存儲(chǔ)空間。兩個(gè)條件滿足任何一個(gè)都會(huì)觸發(fā)清理刪除，如果寫(xiě)入較多導(dǎo)致超過(guò)30%的存儲(chǔ)空間，可能導(dǎo)致Binlog清理而不可用。	需要調(diào)整RDS MySQL的Binlog過(guò)期策略，使得Binlog能正常被讀取。
  通過(guò)只讀實(shí)例消費(fèi)CDC數(shù)據(jù)時(shí)，RDS的只讀實(shí)例不保證Binlog可用（本地最短可能只保留10秒，然后上傳OSS）。如果配置讀取只讀實(shí)例，一旦作業(yè)Failover 10s內(nèi)無(wú)法恢復(fù)，就會(huì)發(fā)生報(bào)錯(cuò)。	不建議MySQL CDC源表讀取RDS的只讀實(shí)例數(shù)據(jù)。 說(shuō)明 是否是只讀實(shí)例可以通過(guò)hostname區(qū)分，hostname為rr開(kāi)頭的是只讀實(shí)例，rm開(kāi)頭的是正常的實(shí)例。
  RDS MySQL發(fā)生了內(nèi)部遷移操作，導(dǎo)致報(bào)錯(cuò)。	需要全新啟動(dòng)作業(yè)，以重新讀取數(shù)據(jù)。

報(bào)錯(cuò)：EventDataDeserializationException: Failed to deserialize data of EventHeaderV4.... Caused by: java.net.SocketException: Connection reset

• 報(bào)錯(cuò)詳情

  EventDataDeserializationException: Failed to deserialize data of EventHeaderV4 .... Caused by: java.net.SocketException: Connection reset
      at com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.deserializeEventData(EventDeserializer.java:304)
      at com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.nextEvent(EventDeserializer.java:227)
      at io.debezium.connector.mysql.BinlogReader$1.nextEvent(BinlogReader.java:252)
      at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:934)
  ... 3 more
  Caused by: java.io.EOFException
      at com.github.shyiko.mysql.binlog.io.ByteArrayInputStream.read (ByteArrayInputStream.java:192)
      at java.io.InputSt ream.read (InputStream.java:170)
      at java.io.InputSt ream.skip (InputStream.java:224)
      at com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.deserializeEventData(EventDeserializer.java:301)
  ...    6 more

• 報(bào)錯(cuò)原因

  • 網(wǎng)絡(luò)問(wèn)題導(dǎo)致。

  • 作業(yè)反壓導(dǎo)致。

    當(dāng)Flink作業(yè)存在反壓時(shí)，CDC Source中使用的Binlog Client會(huì)因?yàn)榉磯旱拇嬖诙鵁o(wú)法繼續(xù)讀取數(shù)據(jù)。為了盡量減少數(shù)據(jù)庫(kù)上殘留的連接數(shù)，MySQL當(dāng)某個(gè)Binlog Client連接不活躍時(shí)間超過(guò)數(shù)據(jù)庫(kù)上配置的超時(shí)時(shí)間之后，會(huì)自動(dòng)切斷連接，導(dǎo)致作業(yè)異常。

  • 數(shù)據(jù)庫(kù)net_write_timeout參數(shù)配置過(guò)小導(dǎo)致。net_write_timeout參數(shù)的默認(rèn)值是60s，如果配置過(guò)小會(huì)導(dǎo)致服務(wù)端主動(dòng)斷開(kāi)鏈接。

• 解決方案

  • 如果是網(wǎng)絡(luò)問(wèn)題導(dǎo)致的，可以嘗試在CDC Source上增加配置'debezium.connect.keep.alive.interval.ms' = '40000'來(lái)解決。如果允許修改數(shù)據(jù)庫(kù)的配置，也可以增加net_write_timeout參數(shù)配置值。具體說(shuō)明請(qǐng)參見(jiàn)參數(shù)調(diào)優(yōu)建議。

  • 如果是作業(yè)反壓?jiǎn)栴}導(dǎo)致的，需要調(diào)節(jié)作業(yè)解決反壓?jiǎn)栴}。

  • 實(shí)時(shí)計(jì)算引擎8.0.7及以上版本增加了該異常的重試，可以嘗試使用實(shí)時(shí)計(jì)算引擎8.0.7及以上版本運(yùn)行作業(yè)。

報(bào)錯(cuò)：The slave is connecting using CHANGE MASTER TO MASTER_AUTO_POSITION = 1, but the master has purged binary logs containing GTIDs that the slave requires.

• 報(bào)錯(cuò)詳情

  org.apache.kafka.connect.errors.ConnectException: The slave is connecting using CHANGE MASTER TO MASTER_AUTO_POSITION = 1, but the master has purged binary logs containing GTIDS that the slave requires. Error code: 1236; SQLSTATE: HY000.
      at io.debezium.connector.mysql.AbstractReader.wrap(AbstractReader.java:241) 
      at io.debezium.connector.mysql.AbstractReader.failed(AbstractReader.java:207) 
      at io.debezium.connector.mysql.BinlogReadersReaderThreadLifecycleListener.onCommunicationFailure(BinlogReader.java:1142) 
      at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:962)
      at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:595)
      at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:839)
      at java.lang.Thread.run(Thread. java:834)
  Caused by: com.github.shyiko.mysql. binlog.network.ServerException: The slave is connecting using CHANGE MASTER TO MASTER_AUTO_POSITION = 1, but the master has purged binary logs containing GTIDs that the slave requires.
      at com.github.shyiko.mysql.binlog. BinaryLogClient.listenForEventPackets(BinaryLogClient.java:926) 
  ... 3 more

• 報(bào)錯(cuò)原因

  全量階段讀取時(shí)間過(guò)長(zhǎng)，導(dǎo)致全量階段結(jié)束開(kāi)始讀Binlog時(shí)，之前記錄的Gtid位點(diǎn)已經(jīng)被MySQL清理掉了。

• 解決方案

  建議延長(zhǎng)Binlog的清理時(shí)間或調(diào)大Binlog文件大小。調(diào)節(jié)清理時(shí)間的命令如下。

  mysql> show variables like 'expire_logs_days';
  mysql> set global expire_logs_days=7;

報(bào)錯(cuò)：java.lang.IllegalStateException: The "before" field of UPDATE/DELETE message is null,please check the Postgres table has been set REPLICA IDENTITY to FULL level.

• 報(bào)錯(cuò)詳情

  java.lang.IllegalStateException: The "before" feild of UPDATE/DELETE message is null,please check the Postgres table has been set REPLICA IDENTITY to FULL level. You can update the setting by running the command in Postgres 'ALTER TABLE xxx.xxx REPLICA IDENTITY FULL'. Please see more in Debezium documentation:https:debezium.io/documentation/reference/1.2/connectors/postresql.html#postgresql-replica-identity
      at com.alibaba.ververica.cdc.connectors.postgres.table.PostgresValueValidator.validate(PostgresValueValidator.java:46)
      at com.alibaba.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema.deserialize(RowDataDebeziumDeserializeSchema.java:113)
      at com.alibaba.ververica.cdc.debezium.internal.DebeziumChangeConsumer.handleBatch(DebeziumChangeConsumer.java:158)
      at io.debezium.embedded.ConvertingEngineBuilder.lambda?notifying$2(ConvertingEngineBuilder.java:82)
      at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:812)
      at io.debezium.embedded.ConvertingEngineBuilder$2.run(ConvertingEngineBuilder.java:171) 
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1147)
      at java.util.concurrent.ThreadPoolExecutorSWorker.run(ThreadPoolExecutor.java:622)
      at java.lang.Thread.run(Thread.java:834)

• 報(bào)錯(cuò)原因

  Postgres表的REPLICA IDENTITY不為FULL時(shí)，會(huì)出現(xiàn)該報(bào)錯(cuò)。

• 解決方案

  根據(jù)提示執(zhí)行ALTER TABLE yourTableName REPLICA IDENTITY FULL;，如果執(zhí)行后作業(yè)重啟依舊報(bào)錯(cuò)，可以嘗試加上'debezium.slot.drop.on.stop' = 'true'參數(shù)解決。

報(bào)錯(cuò)：Caused by: java.lang.IllegalArgumentException: Can't find any matched tables, please check your configured database-name: xxx and table-name: xxxx

• 報(bào)錯(cuò)原因

  • 配置的表名無(wú)法在數(shù)據(jù)庫(kù)中找到時(shí)，出現(xiàn)該報(bào)錯(cuò)。

  • Flink作業(yè)里包含不同數(shù)據(jù)庫(kù)的表，當(dāng)使用的賬號(hào)沒(méi)有其中某些數(shù)據(jù)庫(kù)權(quán)限時(shí)，出現(xiàn)該報(bào)錯(cuò)。

• 解決方案

  1. 檢查對(duì)應(yīng)表名是否在數(shù)據(jù)庫(kù)中存在。

  2. 為賬號(hào)添加作業(yè)中所有數(shù)據(jù)庫(kù)的對(duì)應(yīng)權(quán)限。

報(bào)錯(cuò)：com.github.shyiko.mysql.binlog.network.ServerException

• 報(bào)錯(cuò)詳情

  Caused by: com.github.shyiko.mysql.binlog.network.ServerException: The slave is connecting using CHANGE MASTER TO MASTER_AUTO_POSITION = 1, but the master has purged binary logs containing GTIDs that the slave requires.

• 報(bào)錯(cuò)原因

  MySQL CDC在啟動(dòng)全量讀取之前記錄Binlog位點(diǎn)，等全量讀取結(jié)束后再?gòu)腂inlog位點(diǎn)讀取增量數(shù)據(jù)。該報(bào)錯(cuò)一般是因?yàn)槿孔x取耗時(shí)太長(zhǎng)，超過(guò)了MySQL Binlog的淘汰周期，導(dǎo)致MySQL Binlog位點(diǎn)的數(shù)據(jù)已經(jīng)被MySQL清理掉了。

• 解決方案

  查看MySQL Binlog的清理規(guī)則，例如時(shí)間、存儲(chǔ)空間和文件個(gè)數(shù)等，建議保留Binlog一天以上，RDS Binlog詳情請(qǐng)參見(jiàn)刪除本地日志（Binlog）。

  說(shuō)明

  ：VVR 4.0.8及以上版本，MySQL CDC支持并發(fā)讀取全量數(shù)據(jù)，可以提高全量數(shù)據(jù)的讀取速度，針對(duì)該問(wèn)題會(huì)起到緩解作用。

報(bào)錯(cuò)：The primary key is necessary when enable 'Key: 'scan.incremental.snapshot.enabled'

• 報(bào)錯(cuò)詳情

  MySQL CDC源表在VVR 4.0.x版本語(yǔ)法檢查時(shí)報(bào)錯(cuò)詳情如下。

  Caused by: org.apache.flink.table.api.ValidationException: The primary key is necessary when enable 'Key: 'scan.incremental.snapshot.enabled' , default: true (fallback keys: [])' to 'true'
      at com.alibaba.ververica.cdc.connectors.mysql.table.MySqlTableSourceFactory.validatePrimaryKeyIfEnableParallel(MySqlTableSourceFactory.java:186)
      at com.alibaba.ververica.cdc.connectors.mysql.table.MySqlTableSourceFactory.createDynamicTableSource(MySqlTableSourceFactory.java:85)
      at org.apache.flink.table.factories.FactoryUtil.createTableSource(FactoryUtil.java:134)
      ... 30 more

• 報(bào)錯(cuò)原因

  在MySQL CDC DDL WITH參數(shù)中，您未設(shè)置主鍵（Primary Key）信息。因?yàn)閂VR 6.0.x版本和VVR 4.0.8及以上版本，新增支持按PK分片，進(jìn)行多并發(fā)讀取數(shù)據(jù)的功能。

  重要

  VVR 4.0.8以下版本，MySQL CDC源表僅支持單并發(fā)讀取數(shù)據(jù)。

• 解決方案

  • VVR 6.0.x版本或VVR 4.0.8及以上版本，如果您需要多并發(fā)讀取MySQL數(shù)據(jù)，可以在DDL中添加PK信息。

  • VVR 4.0.8以下版本，MySQL CDC源表不支持多并發(fā)讀取MySQL數(shù)據(jù)，需要在DDL中添加scan.incremental.snapshot.enabled參數(shù)，且把該參數(shù)值設(shè)置為false，無(wú)需設(shè)置PK信息。

報(bào)錯(cuò)：java.io.EOFException: SSL peer shut down incorrectly

• 報(bào)錯(cuò)詳情

  Caused by: java.io.EOFException: SSL peer shut down incorrectly
      at sun.security.ssl.SSLSocketInputRecord.decodeInputRecord(SSLSocketInputRecord.java:239) ~[?:1.8.0_302]
      at sun.security.ssl.SSLSocketInputRecord.decode(SSLSocketInputRecord.java:190) ~[?:1.8.0_302]
      at sun.security.ssl.SSLTransport.decode(SSLTransport.java:109) ~[?:1.8.0_302]
      at sun.security.ssl.SSLSocketImpl.decode(SSLSocketImpl.java:1392) ~[?:1.8.0_302]
      at sun.security.ssl.SSLSocketImpl.readHandshakeRecord(SSLSocketImpl.java:1300) ~[?:1.8.0_302]
      at sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:435) ~[?:1.8.0_302]
      at com.mysql.cj.protocol.ExportControlled.performTlsHandshake(ExportControlled.java:347) ~[?:?]
      at com.mysql.cj.protocol.StandardSocketFactory.performTlsHandshake(StandardSocketFactory.java:194) ~[?:?]
      at com.mysql.cj.protocol.a.NativeSocketConnection.performTlsHandshake(NativeSocketConnection.java:101) ~[?:?]
      at com.mysql.cj.protocol.a.NativeProtocol.negotiateSSLConnection(NativeProtocol.java:308) ~[?:?]
      at com.mysql.cj.protocol.a.NativeAuthenticationProvider.connect(NativeAuthenticationProvider.java:204) ~[?:?]
      at com.mysql.cj.protocol.a.NativeProtocol.connect(NativeProtocol.java:1369) ~[?:?]
      at com.mysql.cj.NativeSession.connect(NativeSession.java:133) ~[?:?]
      at com.mysql.cj.jdbc.ConnectionImpl.connectOneTryOnly(ConnectionImpl.java:949) ~[?:?]
      at com.mysql.cj.jdbc.ConnectionImpl.createNewIO(ConnectionImpl.java:819) ~[?:?]
      at com.mysql.cj.jdbc.ConnectionImpl.<init>(ConnectionImpl.java:449) ~[?:?]
      at com.mysql.cj.jdbc.ConnectionImpl.getInstance(ConnectionImpl.java:242) ~[?:?]
      at com.mysql.cj.jdbc.NonRegisteringDriver.connect(NonRegisteringDriver.java:198) ~[?:?]
      at org.apache.flink.connector.jdbc.internal.connection.SimpleJdbcConnectionProvider.getOrEstablishConnection(SimpleJdbcConnectionProvider.java:128) ~[?:?]
      at org.apache.flink.connector.jdbc.internal.AbstractJdbcOutputFormat.open(AbstractJdbcOutputFormat.java:54) ~[?:?]
      ... 14 more

• 報(bào)錯(cuò)原因

  在MySQL 8.0.27版本，MySQL數(shù)據(jù)庫(kù)默認(rèn)連接需要使用SSL協(xié)議，但JDBC默認(rèn)的訪問(wèn)方式不通過(guò)SSL協(xié)議連接數(shù)據(jù)庫(kù)，導(dǎo)致報(bào)錯(cuò)。

• 解決方案

  • 如果可以升級(jí)到VVR 6.0.2及以上版本，在MySQL CDC表的with參數(shù)中添加參數(shù)'jdbc.properties.useSSL'='false'可以解決該問(wèn)題。

  • 如果聲明的表只是做維表，可以在MySQL表的WITH參數(shù)中將connector設(shè)置為rds，同時(shí)在URL參數(shù)中追加characterEncoding=utf-8&useSSL=false，例如：

    'url'='jdbc:mysql://***.***.***.***:3306/test?characterEncoding=utf-8&useSSL=false'

報(bào)錯(cuò)：com.github.shyiko.mysql.binlog.network.ServerException: A slave with the same server_uuid/server_id as this slave has connected to the master

• 報(bào)錯(cuò)詳情

  Caused by: com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.errors.ConnectException: An exception occurred in the change event producer. This connector will be stopped.
      at io.debezium.pipeline.ErrorHandler.setProducerThrowable(ErrorHandler.java:42) ~[?:?]
      at io.debezium.connector.mysql.MySqlStreamingChangeEventSource$ReaderThreadLifecycleListener.onCommunicationFailure(MySqlStreamingChangeEventSource.java:1185) ~[?:?]
      at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:973) ~[?:?]
      at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:606) ~[?:?]
      at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:850) ~[?:?]
      ... 1 more
  Caused by: io.debezium.DebeziumException: A slave with the same server_uuid/server_id as this slave has connected to the master; the first event '' at 4, the last event read from '/home/mysql/dataxxx/mysql/mysql-bin.xxx' at xxx, the last byte read from '/home/mysql/dataxxx/mysql/mysql-bin.xxx' at xxx. Error code: 1236; SQLSTATE: HY000.
      at io.debezium.connector.mysql.MySqlStreamingChangeEventSource.wrap(MySqlStreamingChangeEventSource.java:1146) ~[?:?]
      at io.debezium.connector.mysql.MySqlStreamingChangeEventSource$ReaderThreadLifecycleListener.onCommunicationFailure(MySqlStreamingChangeEventSource.java:1185) ~[?:?]
      at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:973) ~[?:?]
      at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:606) ~[?:?]
      at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:850) ~[?:?]
      ... 1 more
  Caused by: com.github.shyiko.mysql.binlog.network.ServerException: A slave with the same server_uuid/server_id as this slave has connected to the master; the first event '' at 4, the last event read from '/home/mysql/dataxxx/mysql/mysql-bin.xxx' at xxx, the last byte read from '/home/mysql/dataxxx/mysql/mysql-bin.xxx' at xxx.
      at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:937) ~[?:?]
      at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:606) ~[?:?]
      at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:850) ~[?:?]
      ... 1 more

• 報(bào)錯(cuò)原因

  MySQL CDC源表讀取數(shù)據(jù)時(shí)，需要保證每個(gè)并行度有單獨(dú)的server-id，且每個(gè)server-id全局唯一。當(dāng)使用的server-id與同一作業(yè)CDC源表、其他作業(yè)CDC源表或其他同步工具使用server-id沖突時(shí)，導(dǎo)致報(bào)錯(cuò)。

• 解決方案

  需要為MySQL CDC源表的每個(gè)并行度配置全局唯一的server-id，詳細(xì)操作請(qǐng)參見(jiàn)MySQL CDC源表注意事項(xiàng)。

報(bào)錯(cuò)：TableMapEventDataDeserializer.readMetadata的NullPointerException

• 報(bào)錯(cuò)詳情

  Causedby:java.lang.NullPointerException
      atcom.github.shyiko.mysql.binlog.event.deserialization.TableMapEventDataDeserializer.readMetadata(TableMapEventDataDeserializer.java:81)
  atcom.github.shyiko.mysql.binlog.event.deserialization.TableMapEventDataDeserializer.deserialize(TableMapEventDataDeserializer.java:42)
  atcom.github.shyiko.mysql.binlog.event.deserialization.TableMapEventDataDeserializer.deserialize(TableMapEventDataDeserializer.java:27)
  atcom.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.deserializeEventData(EventDeserializer.java:303)
  atcom.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.deserializeTableMapEventData(EventDeserializer.java:281)
  atcom.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.nextEvent(EventDeserializer.java:228)
  atio.debezium.connector.mysql.MySqlStreamingChangeEventSource$1.nextEvent(MySqlStreamingChangeEventSource.java:259)
  atcom.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:952)
  ...3more

• 報(bào)錯(cuò)原因

  8.0.18及以上版本MySQL新添加了一些數(shù)據(jù)類型，但是Flink解析Binlog部分沒(méi)有兼容新的數(shù)據(jù)類型。

• 解決方案

  VVR 6.0.6及以上版本針對(duì)MySQL新增的數(shù)據(jù)類型進(jìn)行了兼容，您可以升級(jí)VVR版本解決。

MySQL全量過(guò)程中增加列后報(bào)錯(cuò)NullPointerException

• 報(bào)錯(cuò)詳情

  Caused by: org.apache.flink.util.FlinkRuntimeException: Read split MySqlSnapshotSplit{tableId=iplus.retail_detail, splitId='iplus.retail_detail:68', splitKeyType=[`id` BIGINT NOT NULL], splitStart=[212974500236****], splitEnd=[213118153601****], highWatermark=null} error due to java.lang.NullPointerException.
    at com.ververica.cdc.connectors.mysql.debezium.reader.SnapshotSplitReader.checkReadException(SnapshotSplitReader.java:361)
    at com.ververica.cdc.connectors.mysql.debezium.reader.SnapshotSplitReader.pollSplitRecords(SnapshotSplitReader.java:293)
    at com.ververica.cdc.connectors.mysql.source.reader.MySqlSplitReader.pollSplitRecords(MySqlSplitReader.java:124)
    at com.ververica.cdc.connectors.mysql.source.reader.MySqlSplitReader.fetch(MySqlSplitReader.java:86)
    at org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:58)
    at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:142)
    ... 6 more

• 報(bào)錯(cuò)原因

  全量階段的表結(jié)構(gòu)是在作業(yè)啟動(dòng)的時(shí)候確定，且Schema會(huì)被記錄在系統(tǒng)檢查點(diǎn)中。如果在讀全量的過(guò)程中增加了列，Schema會(huì)無(wú)法匹配，就會(huì)拋出此異常。

• 解決方案

  停止作業(yè)并刪除同步的下游表后，無(wú)狀態(tài)啟動(dòng)該作業(yè)。

報(bào)錯(cuò) ：The connector is trying to read binlog starting at GTIDs xxx and binlog file 'binlog.000064', pos=89887992, skipping 4 events plus 1 rows, but this is no longer available on the server. Reconfigure the connector to use a snapshot when needed

• 報(bào)錯(cuò)詳情

  出現(xiàn)該報(bào)錯(cuò)表明CDC作業(yè)試圖讀取的Binlog文件時(shí)，在MySQL服務(wù)器上文件已經(jīng)被清理掉。

• 報(bào)錯(cuò)原因

  • MySQL服務(wù)器上設(shè)置的Binlog文件過(guò)期時(shí)間太短導(dǎo)致文件被自動(dòng)清理。

  • CDC作業(yè)處理Binlog的速度過(guò)慢。

• 解決方案

  • 增加Binlog的保留時(shí)間可以避免文件過(guò)期時(shí)間太短這個(gè)問(wèn)題，例如，將其設(shè)置為保留7天。

    mysql> show variables like 'expire_logs_days';
    mysql> set global expire_logs_days=7;

  • 針對(duì)作業(yè)處理Binlog的速度過(guò)慢這個(gè)問(wèn)題，可以分配更多資源給Flink作業(yè)幫助加快處理速度。

報(bào)錯(cuò)：Mysql8.0 Public Key Retrieval is not allowed

• 報(bào)錯(cuò)原因

  因?yàn)橛脩襞渲玫腗ySQL用戶使用的是SHA256密碼認(rèn)證方式，并且需要TLS等協(xié)議傳輸密碼。

• 解決方案

  允許MySQL用戶支持原始密碼方式訪問(wèn)。更改驗(yàn)證方式的命令如下。

  mysql> ALTER USER 'username'@'localhost' IDENTIFIED WITH mysql_native_password BY 'password';
  mysql> FLUSH PRIVILEGES;