[HUDI-6313] Add metrics counters for compaction requested/completed events.

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/HoodieMetrics.java

@@ -50,6 +50,8 @@ public class HoodieMetrics {
   private String conflictResolutionTimerName = null;
   private String conflictResolutionSuccessCounterName = null;
   private String conflictResolutionFailureCounterName = null;
+  private String compactionStartCounterName = null;
+  private String compactionStopCounterName = null;
   private HoodieWriteConfig config;
   private String tableName;
   private Timer rollbackTimer = null;
@@ -64,6 +66,8 @@ public class HoodieMetrics {
   private Timer conflictResolutionTimer = null;
   private Counter conflictResolutionSuccessCounter = null;
   private Counter conflictResolutionFailureCounter = null;
+  private Counter compactionStartCounter = null;
+  private Counter compactionStopCounter = null;
 
   public HoodieMetrics(HoodieWriteConfig config) {
     this.config = config;
@@ -82,6 +86,8 @@ public HoodieMetrics(HoodieWriteConfig config) {
       this.conflictResolutionTimerName = getMetricsName("timer", "conflict_resolution");
       this.conflictResolutionSuccessCounterName = getMetricsName("counter", "conflict_resolution.success");
       this.conflictResolutionFailureCounterName = getMetricsName("counter", "conflict_resolution.failure");
+      this.compactionStartCounterName = getMetricsName("counter", "compaction.start");
+      this.compactionStopCounterName = getMetricsName("counter", "compaction.stop");
     }
   }
 
@@ -297,6 +303,20 @@ public void emitConflictResolutionFailed() {
     }
   }
 
+  public void emitCompactionStart() {
+    if (config.isMetricsOn()) {
+      compactionStartCounter = getCounter(compactionStartCounter, compactionStartCounterName);
+      compactionStartCounter.inc();
+    }
+  }
+
+  public void emitCompactionStop() {
+    if (config.isMetricsOn()) {
+      compactionStopCounter = getCounter(compactionStopCounter, compactionStopCounterName);
+      compactionStopCounter.inc();
+    }
+  }
+
   private Counter getCounter(Counter counter, String name) {
     if (counter == null) {
       return metrics.getRegistry().counter(name);

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/RunCompactionActionExecutor.java

@@ -35,11 +35,15 @@
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieCompactionException;
 import org.apache.hudi.internal.schema.utils.SerDeHelper;
+import org.apache.hudi.metrics.HoodieMetrics;
 import org.apache.hudi.table.HoodieCompactionHandler;
 import org.apache.hudi.table.HoodieTable;
 import org.apache.hudi.table.action.BaseActionExecutor;
 import org.apache.hudi.table.action.HoodieWriteMetadata;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.util.List;
 
 import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
@@ -48,10 +52,14 @@
 public class RunCompactionActionExecutor<T> extends
     BaseActionExecutor<T, HoodieData<HoodieRecord<T>>, HoodieData<HoodieKey>, HoodieData<WriteStatus>, HoodieWriteMetadata<HoodieData<WriteStatus>>> {
 
+  private static final Logger LOG = LoggerFactory.getLogger(LogCompactionExecutionHelper.class);
+
   private final HoodieCompactor compactor;
   private final HoodieCompactionHandler compactionHandler;
   private WriteOperationType operationType;
 
+  private final HoodieMetrics metrics;
+
   public RunCompactionActionExecutor(HoodieEngineContext context,
                                      HoodieWriteConfig config,
                                      HoodieTable table,
@@ -65,10 +73,14 @@ public RunCompactionActionExecutor(HoodieEngineContext context,
     this.operationType = operationType;
     checkArgument(operationType == WriteOperationType.COMPACT || operationType == WriteOperationType.LOG_COMPACT,
         "Only COMPACT and LOG_COMPACT is supported");
+    metrics = new HoodieMetrics(config);
   }
 
   @Override
   public HoodieWriteMetadata<HoodieData<WriteStatus>> execute() {
+    LOG.info("Compaction start.");
+    metrics.emitCompactionStart();
+
     HoodieTimeline pendingMajorOrMinorCompactionTimeline = WriteOperationType.COMPACT.equals(operationType)
         ? table.getActiveTimeline().filterPendingCompactionTimeline()
         : table.getActiveTimeline().filterPendingLogCompactionTimeline();
@@ -116,6 +128,8 @@ public HoodieWriteMetadata<HoodieData<WriteStatus>> execute() {
       throw new HoodieCompactionException("Could not compact " + config.getBasePath(), e);
     }
 
+    LOG.info("Compaction stop.");
+    metrics.emitCompactionStop();
     return compactionMetadata;
   }
 }