[TRACE] org.apache.hadoop.hbase.regionserver.RegionMergeTransaction

在之前的文章中, 我們介紹了org.apache.hadoop.hbase.regionserver,
在org.apache.hadoop.hbase.regionserver這個函式中,
使用RegionMergeTransaction物件執行Region的合併:

RegionMergeTransaction rmt = new RegionMergeTransaction(a, b, true);

其中, a, b 是要被合併的Region.
rmt則是合併之後, 新展生的Region.
為了詳細了解Region在進行合併時的操作,
我們將繼續研讀RegionMergeTransaction的原始碼,

首先, 我們先來看RegionMergeTransaction的state machine:

    /**
     * Types to add to the transaction journal. Each enum is a step in the merge
     * transaction. Used to figure how much we need to rollback.
     */
    enum JournalEntry {
      /**
       * Set region as in transition, set it into MERGING state.
       */
      SET_MERGING,
      /**
       * We created the temporary merge data directory.
       */
     CREATED_MERGE_DIR,
     /**
      * Closed the merging region A.
      */
     CLOSED_REGION_A,
     /**
      * The merging region A has been taken out of the server's online regions list.
      */
     OFFLINED_REGION_A,
     /**
      * Closed the merging region B.
      */
     CLOSED_REGION_B,
     /**
      * The merging region B has been taken out of the server's online regions list.
      */
     OFFLINED_REGION_B,
     /**
      * Started in on creation of the merged region.
      */
     STARTED_MERGED_REGION_CREATION,
     /**
      * Point of no return. If we got here, then transaction is not recoverable
      * other than by crashing out the regionserver.
      */
     PONR
   }

在RegionMergeTransaction的過程中, 先把a, b兩個Region設成merging的狀態,
之後, 開啟一個新的Region作為合併後的Region,
關閉(close)並移除(offline)被合併的Region a和Region b,
最後再開啟合併後的Region.

在這裡, 我們直接跳掉RegionMergeTransaction檢查條件的部分,
(在perpare()中, 檢查是否屬於同一張表, 是否相鄰, Region狀態等...)
以下是執行的主程式:

   /**
    * Run the transaction.
    * @param server Hosting server instance. Can be null when testing
    * @param services Used to online/offline regions.
    * @throws IOException If thrown, transaction failed. Call
    *           {@link #rollback(Server, RegionServerServices)}
    * @return merged region
    * @throws IOException
    * @see #rollback(Server, RegionServerServices)
    */
   public HRegion execute(final Server server,
  final RegionServerServices services) throws IOException {
     if (rsCoprocessorHost == null) {
       rsCoprocessorHost = server != null ?
         ((HRegionServer) server).getRegionServerCoprocessorHost() : null;
     }
     HRegion mergedRegion = createMergedRegion(server, services);
     if (rsCoprocessorHost != null) {
       rsCoprocessorHost.postMergeCommit(this.region_a, this.region_b, mergedRegion);
     }
     return stepsAfterPONR(server, services, mergedRegion);
   }

在程式中, 先取得RegionServer的資訊, 呼叫createMergedRegion(),
在createMergedRegion()中, 則呼叫stepsBeforePONR(server, services, testing),
完成後, 則回到execute()中, 執行stepsAfterPONR(server, services, mergedRegion),
其中, PONR(Point of no return)就是上述state machine是最後的狀態.
經過PONR後, 就無法回到merge前的狀態,

   /**
    * Prepare the merged region and region files.
    * @param server Hosting server instance. Can be null when testing
    * @param services Used to online/offline regions.
    * @return merged region
    * @throws IOException If thrown, transaction failed. Call
    *           {@link #rollback(Server, RegionServerServices)}
    */
   HRegion createMergedRegion(final Server server,
       final RegionServerServices services) throws IOException {
     LOG.info("Starting merge of " + region_a + " and "
         + region_b.getRegionNameAsString() + ", forcible=" + forcible);
     if ((server != null && server.isStopped())
         || (services != null && services.isStopping())) {
       throw new IOException("Server is stopped or stopping");
     }
 
     if (rsCoprocessorHost != null) {
       if (rsCoprocessorHost.preMerge(this.region_a, this.region_b)) {
         throw new IOException("Coprocessor bypassing regions " + this.region_a + " "
             + this.region_b + " merge.");
       }
     }
 
     // If true, no cluster to write meta edits to or to use coordination.
     boolean testing = server == null ? true : server.getConfiguration()
         .getBoolean("hbase.testing.nocluster", false);
 
     HRegion mergedRegion = stepsBeforePONR(server, services, testing);
 
     @MetaMutationAnnotation
     List<Mutation> metaEntries = new ArrayList<Mutation>();
     if (rsCoprocessorHost != null) {
       if (rsCoprocessorHost.preMergeCommit(this.region_a, this.region_b, metaEntries)) {
         throw new IOException("Coprocessor bypassing regions " + this.region_a + " "
             + this.region_b + " merge.");
       }
       try {
         for (Mutation p : metaEntries) {
           HRegionInfo.parseRegionName(p.getRow());
         }
       } catch (IOException e) {
         LOG.error("Row key of mutation from coprocessor is not parsable as region name."
             + "Mutations from coprocessor should only be for hbase:meta table.", e);
         throw e;
       }
     }
 
     // This is the point of no return. Similar with SplitTransaction.
     // IF we reach the PONR then subsequent failures need to crash out this
     // regionserver
     this.journal.add(JournalEntry.PONR);
 
     // Add merged region and delete region_a and region_b
     // as an atomic update. See HBASE-7721. This update to hbase:meta makes the region
     // will determine whether the region is merged or not in case of failures.
     // If it is successful, master will roll-forward, if not, master will
     // rollback
     if (services != null && !services.reportRegionStateTransition(TransitionCode.MERGE_PONR,
         mergedRegionInfo, region_a.getRegionInfo(), region_b.getRegionInfo())) {
       // Passed PONR, let SSH clean it up
       throw new IOException("Failed to notify master that merge passed PONR: "
         + region_a.getRegionInfo().getRegionNameAsString() + " and "
         + region_b.getRegionInfo().getRegionNameAsString());
     }
     return mergedRegion;
   }

在stepsBeforePONR中,  將先通知Master,
把要進行合併的Region a和Region b狀態改變 (READY_TO_MERGE),
之後, 在Region a所在的FileSystem中, 開啟merge的資料路徑,
關閉並且移除Region a和Region b, 並且合併他們的StoreFile,
設定reference files,
最後透過createMergedRegionFromMerges開啟合併後的Region,
在這裡, createMergedRegionFromMerges可以參考HRegion的介紹,

   public HRegion stepsBeforePONR(final Server server, final RegionServerServices services,
       boolean testing) throws IOException {
     if (services != null && !services.reportRegionStateTransition(TransitionCode.READY_TO_MERGE,
         mergedRegionInfo, region_a.getRegionInfo(), region_b.getRegionInfo())) {
       throw new IOException("Failed to get ok from master to merge "
         + region_a.getRegionInfo().getRegionNameAsString() + " and "
         + region_b.getRegionInfo().getRegionNameAsString());
     }
     this.journal.add(JournalEntry.SET_MERGING);
 
     this.region_a.getRegionFileSystem().createMergesDir();
     this.journal.add(JournalEntry.CREATED_MERGE_DIR);
 
     Map<byte[], List<StoreFile>> hstoreFilesOfRegionA = closeAndOfflineRegion(
         services, this.region_a, true, testing);
     Map<byte[], List<StoreFile>> hstoreFilesOfRegionB = closeAndOfflineRegion(
         services, this.region_b, false, testing);
 
     assert hstoreFilesOfRegionA != null && hstoreFilesOfRegionB != null;
 
 
     //
     // mergeStoreFiles creates merged region dirs under the region_a merges dir
     // Nothing to unroll here if failure -- clean up of CREATE_MERGE_DIR will
     // clean this up.
     mergeStoreFiles(hstoreFilesOfRegionA, hstoreFilesOfRegionB);
 
     // Log to the journal that we are creating merged region. We could fail
     // halfway through. If we do, we could have left
     // stuff in fs that needs cleanup -- a storefile or two. Thats why we
     // add entry to journal BEFORE rather than AFTER the change.
     this.journal.add(JournalEntry.STARTED_MERGED_REGION_CREATION);
     HRegion mergedRegion = createMergedRegionFromMerges(this.region_a,
         this.region_b, this.mergedRegionInfo);
     return mergedRegion;
   }

在完成stepsBeforePONR後, 則進入stepsAfterPONR,
這個函式只是簡單的把Region打開(open)並開啟運作(online),
通知Master, 把已經完成合併的Region a和Region b狀態改變 (mergedRegion),

   public HRegion stepsAfterPONR(final Server server, final RegionServerServices services,
       HRegion mergedRegion) throws IOException {
     openMergedRegion(server, services, mergedRegion);
     if (rsCoprocessorHost != null) {
       rsCoprocessorHost.postMerge(this.region_a, this.region_b, mergedRegion);
     }
     return mergedRegion;
   }

參考資料:
https://hbase.apache.org/xref/org/apache/hadoop/hbase/regionserver/RegionMergeTransaction.html
https://hbase.apache.org/apidocs/org/apache/hadoop/hbase/regionserver/RegionMergeTransaction.html
https://hbase.apache.org/xref/org/apache/hadoop/hbase/regionserver/HRegion.html
https://hbase.apache.org/apidocs/org/apache/hadoop/hbase/regionserver/HRegion.html