Hologres：通過JDBC消費Hologres Binlog

前提條件

• 需提前開啟和配置Hologres Binlog，詳情請參見訂閱Hologres Binlog。

• 需建立hg_binlog extension。

  • Hologres V2.0版本前，需要執行個體的Superuser執行以下語句建立Extension才可以使用該功能，Extension針對整個DB生效，一個DB只需執行一次，建立DB需要再次執行。

    --建立
    CREATE extension hg_binlog;
    
    --刪除
    DROP extension hg_binlog;

    重要

    不推薦使用DROP EXTENSION <extension_name> CASCADE;命令級聯卸載Extension。CASCADE（級聯）刪除命令不僅會刪除指定擴充本身，還會一併清除擴充資料（例如PostGIS資料、RoaringBitmap資料、Proxima資料、Binlog資料、BSI資料等）以及依賴該擴充的對象（包括中繼資料、表、視圖、Server資料等）。

  • Hologres從 V2.0版本起，無需手動建立extension即可使用。

• Hologres V2.1版本起，支援通過如下兩種方式進行Binlog消費。

  • 全部版本支援：完成準備工作，包括為目標表建立Publication、為Publication建立Replication Slot後，直接進行目標表的Binlog消費。

    說明

    該方法需授予使用者如下許可權其中之一：

    • 執行個體的Superuser許可權

    • 目標表的Owner許可權、CREATE DATABASE許可權及執行個體的Replication Role許可權。

  • 僅Hologres V2.1版本起支援：為使用者授予目標表的讀許可權，然後進行目標表的Binlog消費。

使用限制

• 僅Hologres V1.1及以上版本支援通過JDBC消費Hologres Binlog，如果您的執行個體是V1.1以下版本，請您使用自助升級或加入HologresDingTalk交流群反饋，詳情請參見如何擷取更多的線上支援？。

• 僅以下資料類型支援消費Hologres Binlog：INTEGER、BIGINT、SMALLINT、TEXT、CHAR(n)、VARCHAR(n)、REAL、DOUBLE PRECISION、BOOLEAN、NUMERIC(38,8)、DATE、TIME、TIMETZ、TIMESTAMP、TIMESTAMPTZ、BYTEA、JSON、SERIAL、OID、int4[]、int8[]、float4[]、float8[]、boolean[]、text[]，從HologresV1.3.36版本開始支援JSONB類型。如果表中有以上類型之外的資料類型，會造成消費失敗。

  說明

  Hologres從V1.3.36版本開始支援JSONB資料類型消費Hologres Binlog，消費之前需要開啟如下GUC參數：

  -- Session層級開啟GUC
  SET hg_experimental_enable_binlog_jsonb = ON;
  
  -- DB層級開啟GUC
  ALTER database <db_name> SET hg_experimental_enable_binlog_jsonb = ON;

• 同普通串連類似，使用JDBC進行Binlog的消費時，所消費的每張表的每個Shard都會使用1個Walsender串連，Walsender串連與普通串連獨立，互不影響。

• Walsenders數也有使用上限，可以通過以下命令查看單個Frontend節點的最大Walsender數（V2.2版本起預設值調整為600，V2.0和V2.1版本預設為1000，V1.1.26至V2.0版本之間預設為100），總的數目需要乘執行個體的Frontend節點數，不同規格執行個體的Frontend節點數請參見執行個體規格概述。

  SHOW max_wal_senders;

  說明

  Hologres執行個體支援的同時消費Binlog的表數量可以通過以下方式計算：表數量<=(max_wal_senders(100或1000) * FrontEnd節點數)/表的Shard Count 。

  例如：

  • 表a和表b的Shard Count都為20，表c的Shard Count為30，則三張表同時進行消費Binlog佔用的Walsenders數量為20 + 20 + 30 = 70。

  • 表a和表b的Shard Count都為20，表a同時有兩個作業在進行Binlog消費，同時進行消費佔用的Walsenders數量為20 *2 + 20 = 60。

  • 一個執行個體有兩個Frontend節點，則其最大Walsenders數為100*2 = 200，最多支援同時消費10張Shard Count為20的表同時進行消費Binlog。

  如果使用JDBC進行Binlog消費的串連數達到上限，會提示FATAL: sorry, too many wal senders already的錯誤資訊，可以按照如下思路進行排查處理：

  1. 檢查使用JDBC進行Binlog消費的作業，減少其中非必要的Binlog消費。

  2. 檢查Table Group與Shard數設計是否合理，詳情請參見Table Group設定最佳實務。

  3. 如串連數仍超出限制，則須考慮擴容執行個體。

• Hologres V2.0.18版本前，唯讀從執行個體不支援通過JDBC消費Binlog功能。從V2.0.18版本起，唯讀從執行個體支援通過JDBC消費Hologres Binlog，但不支援記錄消費進度。

注意事項

Hologres執行個體版本和Flink引擎版本不同支援消費Binlog的方式也不同，說明如下：

準備工作：建立Publication和Replication Slot

Hologres V2.1版本前，需要先為目標表建立Publication、為Publication建立Replication Slot後，才可以進行Binlog消費。

Hologres V2.1版本起，除上述方法外，還支援僅有目標表讀許可權的使用者進行Binlog消費。該方法無法查詢Hologres側記錄的Binlog消費進度，建議消費端自行記錄消費進度。

-- 使用superuser將普通使用者佈建為replication role：
ALTER role <user_name> replication;

-- 使用superuser將replication role設定回普通使用者：
ALTER role <user_name> noreplication;

使用JDBC消費Binlog

1. 添加POM依賴

   使用如下語句添加POM依賴。

   說明

   添加POM依賴，請使用42.2.18及以上版本的JDBC。

           <dependency>
               <groupId>org.postgresql</groupId>
               <artifactId>postgresql</artifactId>
               <version>42.3.8</version>
           </dependency>
           <!-- 用於擷取表schema以及解析binlog -->
           <dependency>
               <groupId>com.alibaba.hologres</groupId>
               <artifactId>holo-client</artifactId>
               <version>2.2.10</version>
   				</dependency>

2. Java程式碼範例

   import com.alibaba.hologres.client.HoloClient;
   import com.alibaba.hologres.client.HoloConfig;
   import com.alibaba.hologres.client.impl.binlog.HoloBinlogDecoder;
   import com.alibaba.hologres.client.model.Record;
   import com.alibaba.hologres.client.model.TableSchema;
   
   import org.postgresql.PGConnection;
   import org.postgresql.PGProperty;
   import org.postgresql.replication.LogSequenceNumber;
   import org.postgresql.replication.PGReplicationStream;
   
   import java.nio.ByteBuffer;
   import java.sql.Connection;
   import java.sql.DriverManager;
   import java.util.Arrays;
   import java.util.List;
   import java.util.Properties;
   
   public class Test {
   
       public static void main (String[] args) throws Exception {
   
           String username = "";
           String password = "";
           String url = "jdbc:postgresql://Endpoint:Port/db_test";
   
           // 建立JDBC串連
           Properties properties = new Properties();
           PGProperty.USER.set(properties, username);
           PGProperty.PASSWORD.set(properties, password);
           PGProperty.ASSUME_MIN_SERVER_VERSION.set(properties, "9.4");
           // 消費Binlog，務必加上以下參數
           PGProperty.REPLICATION.set(properties, "database");
           try (Connection connection = DriverManager.getConnection(url, properties)) {
               // 建立PGReplicationStream並綁定Replicaiton slot，需要指定shardId
               int shardId = 0;
               PGConnection pgConnection = connection.unwrap(PGConnection.class);
               PGReplicationStream pgReplicationStream = pgConnection.getReplicationAPI().replicationStream()
                       .logical()
           	  // 2.1版本起，此處有兩種可行的寫法
               // 方法1：withSlotName參數為準備階段建立的Replication Slot名，不需填寫withSlotOption("table_name","xxx") 
               // 方法2：不需填寫withSlotName參數，需填寫withSlotOption("table_name","xxx") 
                       .withSlotName("slot_name")
                       .withSlotOption("table_name","public.test_messsage_src") // 消費的表名
                       .withSlotOption("parallel_index", shardId)
                       .withSlotOption("batch_size", "1024")
                       .withSlotOption("start_time", "2021-01-01 00:00:00")
                       .withSlotOption("start_lsn","0")
                       .start();
   	
               // 儘管我們不直接使用holo-client的介面消費binlog，但是需要holo-client的介面去解析消費到的資料。        
               // 建立holo-client
               HoloConfig holoConfig = new HoloConfig();
               holoConfig.setJdbcUrl(url);
               holoConfig.setUsername(username);
               holoConfig.setPassword(password);
               HoloClient client = new HoloClient(holoConfig);
   
               // 建立Binlog decoder用於Decode binary資料，schema需要通過HoloClient擷取
               TableSchema schema = client.getTableSchema("test_message_src", true);
               HoloBinlogDecoder decoder = new HoloBinlogDecoder(schema);
   
               // 用於記錄當前消費位點，用以在消費中斷後，用該值進行繼續消費
               Long currentLsn = 0;
               // 消費資料
               ByteBuffer byteBuffer = pgReplicationStream.readPending();
               while (true) {
                   if (byteBuffer != null) {
                       List<BinlogRecord> records = decoder.decode(shardId, byteBuffer);
                       Long latestLsn = 0L;
                       for (BinlogRecord record : records) {
                           latestLsn = record.getBinlogLsn();
                           // Do Something
                           System.out.println( "lsn: " + latestLsn + ", record: " + Arrays.toString(record.getValues()));
                       }
                       // 儲存消費位點
                       currentLsn = latestLsn;                    
                       pgReplicationStream.forceUpdateStatus();
                   }
                   byteBuffer = pgReplicationStream.readPending();
               }
           }
           // pgReplicationStream.close();
           // connection.close();
       }

   建立PGReplicationStream時，需要通過withSlotName指定Replication Slot：

   • Hologres V2.1版本前，需要填寫已建立的Replication Slot名稱。

   • Hologres V2.1版本起，無需填寫withSlotName，只需在Slot Options中指定目標表名。

   此外，withSlotOption可以指定如下參數。

   參數	是否必須	說明
   table_name	當不指定withSlotName時必須，否則為無效參數。	當不指定withSlotName時，table_name代表了想要消費的目標表名。格式為schema_name.table_name或table_name。
   parallel_index	是	在使用PGReplicationStream進行Binlog消費時，一個PGReplicationStream會建立1個Walsender串連，對目標表的1個Shard的Binlog進行消費。parallel_index即代表了消費目標表的第parallel_index個Shard的資料。 假設某個表有3個Shard，則通過Replication Slot消費Binlog所需要的並發數為3，則使用者最多可以建立3個PGReplicationStream，每個PGReplicationStream的parallel_index參數分別是0、1、2。 當前JDBC消費Hologres Binlog並不支援類似Kafka Consumer Group的實現，所以需要使用者自行建立多個PGReplicationStream。
   start_time	否	表示從某個時間點位開始消費Binlog，樣本參數格式為：2021-01-01 12:00:00+08。 如果未指定start_lsn或start_time，分為如下三種情況： 第一次開始消費Replication Slot的Binlog，則從頭開始消費，類似Kafka的Oldest。 曾經消費過Replication Slot的Binlog，則嘗試從之前Commit過的點位開始消費。 對於不指定withSlotName但指定了table_name的使用情境，不論是否曾經消費過該表的Binlog，都會從頭開始消費。
   start_lsn	否	表示從某個lsn之後開始消費Binlog，同時設定優先權高於start_time。
   batch_size	否	單次擷取的Binlog最大批大小，單位為行，預設值為1024。

   說明

   • BinlogRecord 是decoder返回的Record類型，可以通過以下介面擷取這條資料對應的binlog系統欄位，詳情見訂閱Hologres Binlog。

     • getBinlogLsn() 擷取binlog的序號。

     • getBinlogTimestamp() 擷取Binlog的系統時間戳。

     • getBinlogEventType() 擷取Binlog的事件類型。

   • 消費Binlog之後，使用者需要手動Commit點位資訊，確保下次Failover能夠恢複。

使用Holo Client消費Binlog

• 消費Hologres Binlog功能已經整合至Holo Client中，您可以通過指定需要消費的物理表，方便地消費所有Shard的Binlog資料。

• 使用Holo Client消費Binlog，需要佔用與物理表shard數（slot並發數）相同的串連數，請保證串連數充足。

• 使用Holo Client消費Binlog過程中，推薦您根據Shard自行儲存消費點位，在由於網路連接失敗等原因導致消費終止時，可以通過消費點位進行恢複，詳情請參見下方程式碼範例。

1. 添加POM依賴

   使用如下語句添加POM依賴。

   說明

   添加POM依賴，推薦使用2.2.10及以上版本的Holo Client，2.2.9及之前的版本存在記憶體泄露的問題。

   <dependency>
     <groupId>com.alibaba.hologres</groupId>
     <artifactId>holo-client</artifactId>
     <version>2.2.10</version>
   </dependency>

2. Java程式碼範例

   import com.alibaba.hologres.client.BinlogShardGroupReader;
   import com.alibaba.hologres.client.Command;
   import com.alibaba.hologres.client.HoloClient;
   import com.alibaba.hologres.client.HoloConfig;
   import com.alibaba.hologres.client.Subscribe;
   import com.alibaba.hologres.client.exception.HoloClientException;
   import com.alibaba.hologres.client.impl.binlog.BinlogOffset;
   import com.alibaba.hologres.client.model.binlog.BinlogHeartBeatRecord;
   import com.alibaba.hologres.client.model.binlog.BinlogRecord;
   
   import java.util.HashMap;
   import java.util.Map;
   
   public class HoloBinlogExample {
   
       public static BinlogShardGroupReader reader;
   
       public static void main(String[] args) throws Exception {
           String username = "";
           String password = "";
           String url = "jdbc:postgresql://ip:port/database";
           String tableName = "test_message_src";
           String slotName = "hg_replication_slot_1";
   
           // 建立client的參數
           HoloConfig holoConfig = new HoloConfig();
           holoConfig.setJdbcUrl(url);
           holoConfig.setUsername(username);
           holoConfig.setPassword(password);
           holoConfig.setBinlogReadBatchSize(128);
           holoConfig.setBinlogIgnoreDelete(true);
           holoConfig.setBinlogIgnoreBeforeUpdate(true);
           holoConfig.setBinlogHeartBeatIntervalMs(5000L);
           HoloClient client = new HoloClient(holoConfig);
   
           // 擷取表的shard數
           int shardCount = Command.getShardCount(client, client.getTableSchema(tableName));
   
           // 使用map儲存每個shard的消費進度, 初始化為0
           Map<Integer, Long> shardIdToLsn = new HashMap<>(shardCount);
           for (int i = 0; i < shardCount; i++) {
               shardIdToLsn.put(i, 0L);
           }
   
           // 消費binlog的請求，2.1版本前tableName和slotname為必要參數，2.1版本起僅需傳入tableName（等價於前文使用的固定slotName“hg_table_name_slot”）。
           // Subscribe有StartTimeBuilder和OffsetBuilder兩種，此處以前者為例
           Subscribe subscribe = Subscribe.newStartTimeBuilder(tableName, slotName)
                   .setBinlogReadStartTime("2021-01-01 12:00:00")
                   .build();
           // 建立binlog reader
           reader = client.binlogSubscribe(subscribe);
   
           BinlogRecord record;
   
           int retryCount = 0;
           long count = 0;
           while(true) {
               try {
                   if (reader.isCanceled()) {
                       // 根據儲存的消費點位重新建立reader
                       reader = client.binlogSubscribe(subscribe);
                   }
                   while ((record = reader.getBinlogRecord()) != null) {
                       // 消費到最新
                       if (record instanceof BinlogHeartBeatRecord) {
                           // do something
                           continue;
                       }
       
                       // 處理讀取到的binlog record，這裡只做列印
                       System.out.println(record);
       
                       // 處理之後儲存消費點位，異常時可以從此點位恢複
                       shardIdToLsn.put(record.getShardId(), record.getBinlogLsn());
                       count++;
                       
                       // 讀取成功，重設重試次數
                       retryCount = 0;
                   }
               } catch (HoloClientException e) {
                   if (++retryCount > 10) {
                       throw new RuntimeException(e);
                   }
                   // 發生異常時推薦列印warn層級日誌
                   System.out.println(String.format("binlog read failed because %s and retry %s times", e.getMessage(), retryCount));
       
                   // 重試期間進行一定時間的等待
                   Thread.sleep(5000L * retryCount);
       
                   // 用OffsetBuilder建立Subscribe，從而為每個shard指定起始消費點位
                   Subscribe.OffsetBuilder subscribeBuilder = Subscribe.newOffsetBuilder(tableName, slotName);
                   for (int i = 0; i < shardCount; i++) {
                       // BinlogOffset通過setSequence指定lsn，通過setTimestamp指定時間，兩者同時指定lsn優先順序大於時間戳記
                       // 這雷根據shardIdToLsn這個Map中儲存的消費進度進行恢複
                       subscribeBuilder.addShardStartOffset(i, new BinlogOffset().setSequence(shardIdToLsn.get(i)));
                   }
                   subscribe = subscribeBuilder.build();
                   // 關閉reader
                   reader.cancel();
               }
           }
       }
   }

   使用Holo Client消費Binlog時可以指定如下參數。

   參數	是否必須	預設值	說明
   binlogReadBatchSize	否	1024	從每個Shard單次擷取的Binlog最大批次大小，單位為行。
   binlogHeartBeatIntervalMs	否	-1	binlogRead發送BinlogHeartBeatRecord的間隔，-1表示不發送。 當Binlog沒有新資料，每間隔binlogHeartBeatIntervalMs會下發一條BinlogHeartBeatRecord，此record的timestamp表示截止到這個時間這個Shard上的資料都已經消費完成。
   binlogIgnoreDelete	否	false	是否忽略Delete類型的Binlog。
   binlogIgnoreBeforeUpdate	否	false	是否忽略BeforeUpdate類型的Binlog。

常見問題

消費Binlog並提交消費進度後，發現表hologres.hg_replication_progress不存在，或表中沒有消費進度資料，可能原因如下：

• 消費時不通過Replication Slot進行，即不指定參數withSlotName，該情境不支援記錄消費進度。

• 使用了唯讀從執行個體，且該DB是第一次被消費Binlog，此時hologres.hg_replication_progress表建立失敗。Hologres V2.0.18版本起已修複，從執行個體可以正常消費Binlog。Hologres V2.0.18版本前，需要使用主執行個體先消費一次Binlog，從執行個體即可正常消費。

• 如果不是上述原因，請加入HologresDingTalk交流群聯絡值班人員處理，詳情請參見如何擷取更多的線上支援？。