1. 問題描述

經過對HBase Indexer一段時間的使用、測試、觀察，發現通過Phoenix端匯入到HBase表的資料與Solr那邊的資料會產生不一致的現象，具體體現為Solr那邊的資料會比HBase表資料少幾千條或者更多。在公司測試環境以及試點專案生產環境下都會出現這個問題。

公司的測試環境如下描述：
1) C_PICRECORD表，擁有兩個二級索引表C_PICRECORD_IDX和C_PICRECORD_IDX_COLLISION
2) 測試資料80+w條
3) psql方式匯入，發現Solr會比HBase源表少幾千條到上萬條

【注意】

這個現象使用了現場的部分資料集並且在HBase表建立了二級索引表的情況下會很容易重現。

2. 發現問題

HBase indexer內建會有一些監控資料，監控SepEvent事件處理次數、Solr索引新增次數、Solr索引刪除次數，同時還會記錄一些操作失敗的次數。通過官方Github上Wiki對監控資料這部分的介紹：
https://github.com/NGDATA/hbase-indexer/wiki/Metrics

我們採用JMX方式來獲取這部分監控資料，具體操作如下：
1) 在CDH Cloudera Manager管理介面裡面配置Key Value Store Indexer服務開啟JMX服務配置：

配置項包括：

-Dcom.sun.management.jmxremote
 
.port=8999
-Dcom.sun.management.jmxremote.authenticate=false
-Dcom.sun.management.jmxremote.ssl=false

配置儲存後重啟Key Value Store Indexer服務，接下來我們就可以通過JConsole連線到這個埠，來獲取HBase Indexer的監控資料了。

2) 本地測試使用JConsole來獲取HBase Indexer相關監控資料，如下圖所示：

輸入需要監控的機器IP及埠（前面的配置採用8999埠），點選“連線”進入監控介面。

在“MBean”面板下我們可以看到有“hbaseindexer”的相關監控資料，根據HBase Indexer官網Wiki介紹，在“DirectSolrInputDocumentWriter”下儲存了HBase Indexer對Solr進行操作的監控資料，比如新增索引次數“Indexer adds”和刪除索引次數“Indexer deletes”，裡面分別記錄了索引新增次數和索引刪除次數。我們正是通過監控這兩個資料，分析出Solr和HBase源表資料產生不一致的原因——部分HBase新增在HBase Indexer被當成“Indexer deletes”操作處理，從而使得Solr的資料少於HBase源表資料。少掉的資料等於“Indexer deletes”裡面記錄的次數，即有：Hbase Row Count - Solr numfound = Indexer deletes count。

3. 除錯並定位問題

針對上面觀察並分析的現象，我們需要定位到Hbase Indexer原始碼進行分析：

【注意】

需要使用CDH5.4裡面的Key Value Store Indexer原始碼，而不是Github上面HBase Indexer的原始碼。當然，兩者的區別僅在於hbase版本的不同。（CDH使用的是自己封裝改造過的HBase版本）

定位到Indexer類的indexRowData方法，這個方法會呼叫calculateIndexUpdates方法，針對我們的配置採用的是RowBasedIndexer（Indexer的子類，並且是內部類）。因此我們定位到RowBasedIndexer的calculateIndexUpdates實現方法上。

@Override
        protected void calculateIndexUpdates(List<RowData> rowDataList, SolrUpdateCollector updateCollector) throws IOException {

            Map<String, RowData> idToRowData = calculateUniqueEvents(rowDataList); // 選取需要處理的行(包括那些要刪除的)

            for (RowData rowData : idToRowData.values()) {
                String tableName = new String(rowData.getTable(), Charsets.UTF_8);

                Result result = rowData.toResult(); // 對於需要delete(delete or delete family)的行,其kvs為空

                if (conf.getRowReadMode() == RowReadMode.DYNAMIC) {
                    if (!mapper.containsRequiredData(result)) {
                        if(log.isDebugEnabled()) {
                            log.debug("Row " + Bytes.toString(rowData.getRow()) + " need to re-read from hbase");
                        }
                        result = readRow(rowData, result);
                    }
                }

                boolean rowDeleted = result.isEmpty(); // 由此判斷是否為delete row

                String documentId;
                if (uniqueKeyFormatter instanceof UniqueTableKeyFormatter) {
                    documentId = ((UniqueTableKeyFormatter) uniqueKeyFormatter).formatRow(rowData.getRow(),
                            rowData.getTable());
                } else {
                    documentId = uniqueKeyFormatter.formatRow(rowData.getRow());
                }

                if (rowDeleted) {
                    // Delete row from Solr as well
                    updateCollector.deleteById(documentId);
                    if (log.isDebugEnabled()) { 
                        log.debug("Row " + Bytes.toString(rowData.getRow()) + ": deleted from Solr, kvs : " + rowData.getKeyValues());
                    }
                } else {
                    IdAddingSolrUpdateWriter idAddingUpdateWriter = new IdAddingSolrUpdateWriter(
                            conf.getUniqueKeyField(),
                            documentId,
                            conf.getTableNameField(),
                            tableName,
                            updateCollector);
                    mapper.map(result, idAddingUpdateWriter);
                }
            }
        }

出現不一致的根源在於

conf.getRowReadMode() == RowReadMode.DYNAMIC

我們所建立的Indexer預設情況下RowReadMode是採用DYNAMIC的，那麼在程式碼裡面會進入這段條件邏輯，這部分程式碼簡單說明如下：

判斷Result裡面是否包含了所有我們需要對映到Solr裡面的列，如否，則需要呼叫readRow方法從HBase裡面獲取我們需要對映的資料（根據rowkey，呼叫HTable.get(Get)）。

出現不一致的現象就在這部分邏輯裡面，我們考慮下面這種情況：

1) HBase RegionServer 將Put操作先寫WAL （這個時候Put還沒儲存到Region）
2) 非同步處理的HBase Indexer獲取到這個WAL日誌，對資料進行處理，進入了我們上面說的這段條件邏輯程式碼，恰巧Result裡面沒有一部分Solr索引列，那麼需要呼叫readRow方法從HBase重新讀取資料，這個時候呼叫HTable.get(Get) 並沒有獲取到資料（Result.isEmpty()為真）
3) HBase RegionServer把Put儲存到Region
4) 那麼對於2) 裡面的HBase Indexer，那條記錄將被當成delelet操作，所以在後面的邏輯將其當成solr delete document的操作

經過以上分析再加上我們通過新增debug日誌進行除錯，驗證了我們上面的猜測。

配置CDH Key Value Store Indexer log4j配置，加入需要DEBUG的類：

log4j.logger.com.ngdata.hbaseindexer.indexer.Indexer$RowBasedIndexer=DEBUG
log4j.logger.com.ngdata.hbaseindexer.indexer.Indexer$ColumnBasedIndexer=DEBUG

我們通過在原始碼裡面新增輸出自定義DEBUG資訊來除錯（需要編譯並替換叢集上面的hbase-indexer-engine-xxx.jar）。

4. 解決問題

根據上面的驗證以及對程式碼的初步瞭解，我們給出以下兩種解決方案：

4.1 方案1

建立hbase indexer的時候，對morphline-hbase-mapper.xml裡面<indexer>節點增加read-row屬性配置，配置read-row=”never”，關於這個引數的說明參考：

配置為never，將不會進入上面提到的條件邏輯程式碼，那麼自然也不會出現不同步的現象。這種解決方案只需要修改建立indexer的配置，而不需要修改HBase Indexer原始碼，優於後面提到的【方案2】

4.2 方案2

在readRow()方法裡面加入retry邏輯，合理的設定重試次數併合理休眠，以此來保證能夠獲取到正確的Result。這種方式相對比較麻煩，首先需要修改原始碼，並進行編譯，替換叢集上相應jar包。而且，重試和休眠會對效能造成一定影響，建議不到萬不得已不要使用這種方式。