消費者取消彙整 - Amazon Kinesis Data Streams
從舊版 KCL 進行遷移KPL 取消彙整所用的 KCL 延伸模組 GetRecords直接使用

本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

消費者取消彙整

自發行版本 1.4.0 起，KCL 支援自動取消彙整 KPL 使用者記錄。在您更新 KCL 後，使用舊版 KCL 撰寫的取用者應用程式將編譯程式碼而不會進行任何修改。不過，如果生產者端使用了 KPL 彙整，則有一項與檢查點作業相關的細微之處須留意：彙整的記錄中所有的子記錄具有相同的序號，所以若您需要區別子記錄，就必須隨檢查點存放額外的資料。這些額外的資料稱為子序號。

從舊版 KCL 進行遷移

您無須更改任何現有的呼叫即可搭配彙整執行檢查點作業。保證您仍能成功擷取存放於 Kinesis Data Streams 的所有記錄。KCL 現提供兩項新的檢查點操作以支援特定使用案例，如下所述。

若您現有的程式碼是在 KPL 支援之前針對 KCL 撰寫，而且以不帶引數的方式呼叫檢查點操作，此即等同於對批次中最後一筆 KPL 使用者記錄的序號執行檢查點作業。如果使用序號字串呼叫檢查點操作，則等同於對批次的指定序號及隱含的子序號 0 (零) 執行檢查點作業。

呼叫新的 KCL 檢查點操作而不帶任何引數如 checkpoint() 語意上等同於對批次中其上次 Record 呼叫的序號及隱含的子序號 0 (零) 執行檢查點作業。

呼叫新的 KCL 檢查點操作如 checkpoint(Record record) 語意上等同於對指定 Record 的序號及隱含的子序號 0 (零) 執行檢查點作業。若 Record 呼叫實際為 UserRecord，則會對 UserRecord 序號和子序號執行檢查點作業。

呼叫新的 KCL 檢查點操作如 checkpoint(String sequenceNumber, long subSequenceNumber) 會對指定的序號及指定的子序號執行明示檢查點作業。

上述任何情況下，當檢查點已存放於 Amazon DynamoDB 檢查點資料表之後，KCL 便能正確地恢復擷取記錄，就算應用程式當機並重新啟動也沒問題。如果序列中包含多筆記錄，則會從序號最近執行過檢查點作業的記錄中為下一個子序號的記錄開始擷取。若最近的檢查點包括前一序號記錄的最新子序號，將從下一個序號的記錄開始擷取。

下一節將討論需要避免略過記錄和重複記錄的消費者對序號與子序號執行檢查點作業的詳細資訊。若停止並重新啟動消費者的記錄處理會略過 (或重複) 記錄無關緊要，您就可以執行現有的程式碼而無須修改。

KPL 取消彙整所用的 KCL 延伸模組

如前所述，KPL 取消彙整涉及子序列檢查點作業。為方便使用子序列檢查點作業，KCL 增加了 UserRecord 類別：

public class UserRecord extends Record {     
    public long getSubSequenceNumber() {
    /* ... */
    }      
    @Override 
    public int hashCode() {
    /* contract-satisfying implementation */ 
    }      
    @Override 
    public boolean equals(Object obj) {
    /* contract-satisfying implementation */ 
    } 
}

現已使用此類別代替 Record。這不會破壞現有的程式碼，因為其為 Record 的子類別。UserRecord 類別同時代表實際的子記錄和未彙整的標準記錄。未彙整的記錄可想像成恰有一筆子記錄的已彙整記錄。

此外，IRecordProcessorCheckpointer 也增加了兩項新的操作：

public void checkpoint(Record record); 
public void checkpoint(String sequenceNumber, long subSequenceNumber);

若要開始使用子序號檢查點作業，您可進行以下轉換。更改以下形式的程式碼：

checkpointer.checkpoint(record.getSequenceNumber());

新形式的程式碼：

checkpointer.checkpoint(record);

建議您使用 checkpoint(Record record) 形式執行子序列檢查點作業。不過，若您已將 sequenceNumbers 存放在字串中用於檢查點作業，則現在亦應存放 subSequenceNumber，如以下範例所示：

String sequenceNumber = record.getSequenceNumber(); 
long subSequenceNumber = ((UserRecord) record).getSubSequenceNumber();  // ... do other processing  
checkpointer.checkpoint(sequenceNumber, subSequenceNumber);

從 Record 轉換為 UserRecord 絕對會成功，因為背後的實作一律使用 UserRecord。除非需要對序號進行算術運算，否則這種方式並不建議。

處理 KPL 使用者記錄時，KCL 會將子序號寫入 Amazon DynamoDB 成為每一列的額外欄位。舊版 KCL 是使用 AFTER_SEQUENCE_NUMBER 在恢復檢查點作業時擷取記錄。目前有 KPL 支援的 KCL 則改用 AT_SEQUENCE_NUMBER。在擷取已對序號執行過檢查點作業的記錄時，會檢查已執行檢查點作業的子序號，且將視需要刪除子記錄 (若已對最後一筆子記錄執行檢查點作業，則可能全部刪除)。同樣地，未彙整的記錄可想像成只有一筆子記錄的已彙整記錄，所以同一套演算法對已彙整和未彙整的記錄都適用。

GetRecords直接使用

您也可以選擇不使用 KCL 而是直接調用 API 操作 GetRecords 來擷取 Kinesis Data Streams 記錄。若要將這些擷取到的記錄解壓縮為原始 KPL 使用者記錄，請呼叫 UserRecord.java 的以下任一項靜態操作：

public static List<Record> deaggregate(List<Record> records)

public static List<UserRecord> deaggregate(List<UserRecord> records, BigInteger startingHashKey, BigInteger endingHashKey)

第一項操作使用 0 的預設值 startingHashKey (零) 以及 2^128 -1 的預設值 endingHashKey。

上述每一項操作都將取消彙總指定的 Kinesis Data Streams 記錄清單，轉成 KPL 使用者記錄清單。傳回的記錄清單將會捨棄顯式雜湊索引鍵或分割區索引鍵落在 startingHashKey (含) 以及 endingHashKey (含) 範圍外的任何 KPL 使用者記錄。