HIVE-29188: [hiveACIDReplication] Add src and tgt commit time in replication metrics for better monitoring (#6071)

shivaom02 · shivamkmr02 · web-flow · commit 4bb08099d91a · 2025-09-16T11:45:05.000+09:00
* Details:
  * Currently when the Hive ACID replication is running we do not have any idea of how the replication is progressing.
  * Even when the replication is not running, we do not have any idea how much the target is behind the src.
  * This commit is to add this info in replication_metrics.

Co-authored-by: shivam02 &lt;shivam.kumar@cloudera.com&gt;
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/load/message/AbortTxnHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/load/message/AbortTxnHandler.java
@@ -18,6 +18,7 @@
 package org.apache.hadoop.hive.ql.parse.repl.load.message;
 
 import org.apache.hadoop.hive.metastore.messaging.AbortTxnMessage;
+import org.apache.hadoop.hive.metastore.utils.StringUtils;
 import org.apache.hadoop.hive.ql.plan.ReplTxnWork;
 import org.apache.hadoop.hive.ql.exec.Task;
 import org.apache.hadoop.hive.ql.exec.TaskFactory;
@@ -27,6 +28,8 @@
 import java.io.Serializable;
 import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
 
 /**
  * AbortTxnHandler
@@ -43,6 +46,17 @@ public List<Task<?>> handle(Context context)
 
     AbortTxnMessage msg = deserializer.getAbortTxnMessage(context.dmd.getPayload());
 
+    // Saving the timestamp of all write abort txn in metric 'progress' to calculate lag between src and tgt
+    List<Long> writeIds = msg.getWriteIds();
+    List<String> databases = Optional.ofNullable(msg.getDbsUpdated())
+                              .orElse(Collections.emptyList())
+                              .stream()
+                              .map(StringUtils::normalizeIdentifier)
+                              .toList();
+    if (databases.contains(context.dbName) && writeIds != null && !writeIds.isEmpty()) {
+      context.getMetricCollector().setSrcTimeInProgress(msg.getTimestamp());
+    }
+
     Task<ReplTxnWork> abortTxnTask = TaskFactory.get(
         new ReplTxnWork(HiveUtils.getReplPolicy(context.dbName), context.dbName, null,
                 msg.getTxnId(), ReplTxnWork.OperationType.REPL_ABORT_TXN, context.eventOnlyReplicationSpec(),
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/load/message/CommitTxnHandler.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/load/message/CommitTxnHandler.java
@@ -20,6 +20,7 @@
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hive.metastore.api.WriteEventInfo;
 import org.apache.hadoop.hive.metastore.messaging.CommitTxnMessage;
+import org.apache.hadoop.hive.metastore.utils.StringUtils;
 import org.apache.hadoop.hive.ql.exec.repl.util.AddDependencyToLeaves;
 import org.apache.hadoop.hive.ql.exec.util.DAGTraversal;
 import org.apache.hadoop.hive.ql.metadata.Table;
@@ -33,6 +34,8 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
 
 /**
  * CommitTxnHandler
@@ -54,6 +57,17 @@ public List<Task<?>> handle(Context context)
     String tableNamePrev = null;
     String tblName = null;
 
+    // Saving the timestamp of all write commit txn in metric 'progress' to calculate lag between src and tgt
+    List<Long> writeIds = msg.getWriteIds();
+    List<String> databases = Optional.ofNullable(msg.getDatabases())
+                              .orElse(Collections.emptyList())
+                              .stream()
+                              .map(StringUtils::normalizeIdentifier)
+                              .toList();
+    if (databases.contains(dbName) && writeIds != null && !writeIds.isEmpty()) {
+      context.getMetricCollector().setSrcTimeInProgress(msg.getTimestamp());
+    }
+
     ReplTxnWork work = new ReplTxnWork(HiveUtils.getReplPolicy(context.dbName), context.dbName,
                                        null, msg.getTxnId(), ReplTxnWork.OperationType.REPL_COMMIT_TXN,
                                         context.eventOnlyReplicationSpec(), context.getDumpDirectory(),
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/metric/ReplicationMetricCollector.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/metric/ReplicationMetricCollector.java
@@ -136,6 +136,22 @@ public void reportFailoverStart(String stageName, Map<String, Long> metricMap,
     }
   }
 
+  public void setSrcTimeInProgress(long endTimeOnSrc) throws SemanticException {
+    if (isEnabled) {
+      LOG.debug("Updating last commit time on src in progress as: {}", endTimeOnSrc);
+      Progress progress = replicationMetric.getProgress();
+      Stage stage = progress.getStageByName("REPL_LOAD");
+      if (stage == null) {
+        return;
+      }
+      stage.setEndTimeOnSrc(endTimeOnSrc);
+      stage.setEndTimeOnTgt(getCurrentTimeInMillis());
+      progress.addStage(stage);
+      replicationMetric.setProgress(progress);
+      metricCollector.addMetric(replicationMetric);
+    }
+  }
+
   public void reportStageEnd(String stageName, Status status, long lastReplId,
       SnapshotUtils.ReplSnapshotCount replSnapshotCount, ReplStatsTracker replStatsTracker) throws SemanticException {
     unRegisterMBeanSafe();
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/metric/event/Stage.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/repl/metric/event/Stage.java
@@ -32,6 +32,8 @@ public class Stage {
   private Status status;
   private long startTime;
   private long endTime;
+  private long endTimeOnSrc;
+  private long endTimeOnTgt;
   private Map<String, Metric> metrics = new HashMap<>();
   private String errorLogPath;
   private SnapshotUtils.ReplSnapshotCount replSnapshotCount = new SnapshotUtils.ReplSnapshotCount();
@@ -58,6 +60,8 @@ public Stage(Stage stage) {
     this.errorLogPath = stage.errorLogPath;
     this.replSnapshotCount = stage.replSnapshotCount;
     this.replStats = stage.replStats;
+    this.endTimeOnSrc = stage.endTimeOnSrc;
+    this.endTimeOnTgt = stage.endTimeOnTgt;
   }
 
   public String getName() {
@@ -92,6 +96,21 @@ public void setEndTime(long endTime) {
     this.endTime = endTime;
   }
 
+  public long getEndTimeOnSrc() {
+    return endTimeOnSrc;
+  }
+
+  public void setEndTimeOnSrc(long endTimeOnSrc) {
+    this.endTimeOnSrc = endTimeOnSrc;
+  }
+
+  public long getEndTimeOnTgt() {
+    return endTimeOnTgt;
+  }
+
+  public void setEndTimeOnTgt(long endTimeOnTgt) {
+    this.endTimeOnTgt = endTimeOnTgt;
+  }
 
   public void addMetric(Metric metric) {
     this.metrics.put(metric.getName(), metric);
diff --git a/ql/src/test/results/clientpositive/llap/replication_metrics_ingest.q.out b/ql/src/test/results/clientpositive/llap/replication_metrics_ingest.q.out
@@ -92,5 +92,5 @@ POSTHOOK: type: QUERY
 POSTHOOK: Input: sys@replication_metrics
 POSTHOOK: Input: sys@replication_metrics_orig
 #### A masked pattern was here ####
-repl1	1	{"dbName":"src","replicationType":"BOOTSTRAP","replicatedDBSizeInKB":0.0,"stagingDir":"dummyDir","lastReplId":0,"failoverMetadataLoc":null,"failoverEventId":0,"failoverEndPoint":null,"failoverType":null}	H4sIAAAAAAAA/22PwQ6CMBBE/2XPHOTKTSsmJojEwskQ02gDJKUl2+2J9N8tEohEb7sz83ayI1gS5CwkwCvGUs4hmqRGBuk+gha9DN4tLbLHsboUs/sHQCq7KbqLQOrXOveSsHtubp2qnJXnaz6BT4coNTHjNH3yZEioZfXRCpX7Q5b+EvGWiH0d6hENZqYpBLWQaKdUBCgHxbUYbGsW9MsID9lZ8LV/A7NIwGISAQAA	{"status":"SUCCESS","stages":[{"name":"REPL_DUMP","status":"SUCCESS","startTime":0,"endTime":0,"metrics":[{"name":"FUNCTIONS","currentCount":0,"totalCount":0},{"name":"TABLES","currentCount":1,"totalCount":1}],"errorLogPath":null,"replSnapshotCount":null,"replStats":null}]}	gzip(json-2.0)
-repl2	1	{"dbName":"destination","replicationType":"BOOTSTRAP","replicatedDBSizeInKB":0.00390625,"stagingDir":"dummyDir","lastReplId":0,"failoverMetadataLoc":null,"failoverEventId":0,"failoverEndPoint":null,"failoverType":null}	H4sIAAAAAAAA/22PwQqDMBBE/yVnD/XqzUYLBbFS9VSkBF1UiImsm5Pk3xu1CtLedmb3zbAzm0iQmVjA8pLzOM+Zt1gtOOs1MyUGcLtnnCXv5BFG2/YPgFT0y+nFY6CaYx6AsK9PWbcy5cX9kS5gbRBBEddG0XpPmoTcpfUOqAivSfxL+GfCt5WrR9SY6DYT1LFAGSk9hjDKXIlx6vSOumgzcARB0KzVTkYgYZP2y7hfpy3EVvYDvpfiNy0BAAA=	{"status":"SUCCESS","stages":[{"name":"REPL_LOAD","status":"SUCCESS","startTime":0,"endTime":0,"metrics":[{"name":"FUNCTIONS","currentCount":0,"totalCount":0},{"name":"TABLES","currentCount":1,"totalCount":1}],"errorLogPath":null,"replSnapshotCount":{"numCreated":0,"numDeleted":0},"replStats":null}]}	gzip(json-2.0)
+repl1	1	{"dbName":"src","replicationType":"BOOTSTRAP","replicatedDBSizeInKB":0.0,"stagingDir":"dummyDir","lastReplId":0,"failoverMetadataLoc":null,"failoverEventId":0,"failoverEndPoint":null,"failoverType":null}	H4sIAAAAAAAA/22PsQ6CMBRF/6Uzg6xsWjExQSC2TIaYBhsgKS15fZ1I/90iwYi6tefe0/c6EYsCnSUJYRWlKWMkmlErA7pNRItBhuyaltn9WF3KJf0jAPJ+ru4iIvXj+1xoBs0W8BZfYJAIfbOZdqpyys9FPj/dOACpkRqnlz4aFGq9+ugt8f0hS3+NeGvEvg47ABjITFsK7EiinVIRATkqpsVoO7OqH0H4sl2Ar/0T5NGeBTQBAAA=	{"status":"SUCCESS","stages":[{"name":"REPL_DUMP","status":"SUCCESS","startTime":0,"endTime":0,"endTimeOnSrc":0,"endTimeOnTgt":0,"metrics":[{"name":"FUNCTIONS","currentCount":0,"totalCount":0},{"name":"TABLES","currentCount":1,"totalCount":1}],"errorLogPath":null,"replSnapshotCount":null,"replStats":null}]}	gzip(json-2.0)
+repl2	1	{"dbName":"destination","replicationType":"BOOTSTRAP","replicatedDBSizeInKB":0.00390625,"stagingDir":"dummyDir","lastReplId":0,"failoverMetadataLoc":null,"failoverEventId":0,"failoverEndPoint":null,"failoverType":null}	H4sIAAAAAAAA/22QQQuDMAyF/0vPHubV21YdDERl1dOQUVzQQW0lpifpf1/VOXDbLe8l30vIxEaSZEcWMVFxngjBgtlqwVu3iWnZg+9dkyK9p/kxXrt/AKTyOY8eAgb68V3nWmCzN8qWFqMHwmez23auMl5e8myObiwiaOLG6nWeDEm1SRd8oPJ4SpNfItwToav9DYgGU9MWkjoWaatUwBAGJbQcxs5sqI+2PUeQBI9ltZcxKFilezP+G+Ma4mr3Atju2TJPAQAA	{"status":"SUCCESS","stages":[{"name":"REPL_LOAD","status":"SUCCESS","startTime":0,"endTime":0,"endTimeOnSrc":0,"endTimeOnTgt":0,"metrics":[{"name":"FUNCTIONS","currentCount":0,"totalCount":0},{"name":"TABLES","currentCount":1,"totalCount":1}],"errorLogPath":null,"replSnapshotCount":{"numCreated":0,"numDeleted":0},"replStats":null}]}	gzip(json-2.0)
