HBASE-22749: Distributed MOB compactions

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterImpl.java

@@ -227,6 +227,10 @@ protected final void writeFileInfo(FixedFileTrailer trailer, DataOutputStream ou
     HFile.updateWriteLatency(System.currentTimeMillis() - startTime);
   }
 
+  public long getPos() throws IOException {
+    return outputStream.getPos();
+
+  }
   /**
    * Checks that the given Cell's key does not violate the key order.
    *

hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java

@@ -395,9 +395,8 @@ public void run() {
   private LogCleaner logCleaner;
   private HFileCleaner hfileCleaner;
   private ReplicationBarrierCleaner replicationBarrierCleaner;
-  private ExpiredMobFileCleanerChore expiredMobFileCleanerChore;
-  private MobCompactionChore mobCompactChore;
-  private MasterMobCompactionThread mobCompactThread;
+  private MobFileCleanerChore mobFileCleanerChore;
+  private MobFileCompactionChore mobFileCompactionChore;
   // used to synchronize the mobCompactionStates
   private final IdLock mobCompactionLock = new IdLock();
   // save the information of mob compactions in tables.
@@ -1300,14 +1299,18 @@ public void updateConfigurationForQuotasObserver(Configuration conf) {
   }
 
   private void initMobCleaner() {
-    this.expiredMobFileCleanerChore = new ExpiredMobFileCleanerChore(this);
-    getChoreService().scheduleChore(expiredMobFileCleanerChore);
+    this.mobFileCleanerChore = new MobFileCleanerChore(this);
+    getChoreService().scheduleChore(mobFileCleanerChore);
 
     int mobCompactionPeriod = conf.getInt(MobConstants.MOB_COMPACTION_CHORE_PERIOD,
-        MobConstants.DEFAULT_MOB_COMPACTION_CHORE_PERIOD);
-    this.mobCompactChore = new MobCompactionChore(this, mobCompactionPeriod);
-    getChoreService().scheduleChore(mobCompactChore);
-    this.mobCompactThread = new MasterMobCompactionThread(this);
+      MobConstants.DEFAULT_MOB_COMPACTION_CHORE_PERIOD);
+
+    if (mobCompactionPeriod > 0) {
+      this.mobFileCompactionChore = new MobFileCompactionChore(this);
+      getChoreService().scheduleChore(mobFileCompactionChore);
+    } else {
+      LOG.info("The period is " + mobCompactionPeriod + " seconds, MobCompactionChore is disabled");
+    }
   }
 
   /**
@@ -1495,9 +1498,7 @@ protected void stopServiceThreads() {
       }
     }
     stopChores();
-    if (this.mobCompactThread != null) {
-      this.mobCompactThread.close();
-    }
+
     super.stopServiceThreads();
     if (cleanerPool != null) {
       cleanerPool.shutdownNow();
@@ -1618,8 +1619,8 @@ private void stopProcedureExecutor() {
   private void stopChores() {
     ChoreService choreService = getChoreService();
     if (choreService != null) {
-      choreService.cancelChore(this.expiredMobFileCleanerChore);
-      choreService.cancelChore(this.mobCompactChore);
+      choreService.cancelChore(this.mobFileCleanerChore);
+      choreService.cancelChore(this.mobFileCompactionChore);
       choreService.cancelChore(this.balancerChore);
       choreService.cancelChore(this.normalizerChore);
       choreService.cancelChore(this.clusterStatusChore);
@@ -3444,17 +3445,6 @@ public void reportMobCompactionEnd(TableName tableName) throws IOException {
     }
   }
 
-  /**
-   * Requests mob compaction.
-   * @param tableName The table the compact.
-   * @param columns The compacted columns.
-   * @param allFiles Whether add all mob files into the compaction.
-   */
-  public void requestMobCompaction(TableName tableName,
-                                   List<ColumnFamilyDescriptor> columns, boolean allFiles) throws IOException {
-    mobCompactThread.requestMobCompaction(conf, fs, tableName, columns, allFiles);
-  }
-
   /**
    * Queries the state of the {@link LoadBalancerTracker}. If the balancer is not initialized,
    * false is returned.