Updated to use System.nanoTime() for lag calculation

Signed-off-by: Vinay Krishna Pudyodu <vinkrish.neo@gmail.com>
vinaykpud · Jan 21, 2025 · dd0406d · dd0406d
1 parent a94240f
commit dd0406d
Show file tree

Hide file tree

Showing 4 changed files with 8 additions and 7 deletions.
diff --git a/server/src/main/java/org/opensearch/index/shard/IndexShard.java b/server/src/main/java/org/opensearch/index/shard/IndexShard.java
@@ -1788,7 +1788,7 @@ ReplicationCheckpoint computeReplicationCheckpoint(SegmentInfos segmentInfos) th
             metadataMap.values().stream().mapToLong(StoreFileMetadata::length).sum(),
             getEngine().config().getCodec().getName(),
             metadataMap,
-            System.currentTimeMillis()
+            System.nanoTime()
         );
         logger.trace("Recomputed ReplicationCheckpoint for shard {}", checkpoint);
         return checkpoint;

diff --git a/server/src/main/java/org/opensearch/indices/replication/SegmentReplicator.java b/server/src/main/java/org/opensearch/indices/replication/SegmentReplicator.java
@@ -32,6 +32,7 @@
 import java.io.IOException;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.TimeUnit;
 
 /**
  * This class is responsible for managing segment replication events on replicas.
@@ -124,7 +125,7 @@ public ReplicationStats getSegmentReplicationStats(ShardId shardId, ReplicationC
 
         final ReplicationCheckpoint lastOnGoingReplicationCheckpoint = this.lastOnGoingReplicationCheckpoint.get(shardId);
         final long replicationLag = lastOnGoingReplicationCheckpoint != null
-            ? System.currentTimeMillis() - lastOnGoingReplicationCheckpoint.getCreatedTimeStamp()
+            ? TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - lastOnGoingReplicationCheckpoint.getCreatedTimeStamp())
             : 0;
 
         return new ReplicationStats(bytesBehindSum, bytesBehindSum, bytesBehindSum > 0L ? replicationLag : 0);

diff --git a/...er/src/main/java/org/opensearch/indices/replication/checkpoint/ReplicationCheckpoint.java b/...er/src/main/java/org/opensearch/indices/replication/checkpoint/ReplicationCheckpoint.java
@@ -56,11 +56,11 @@ private ReplicationCheckpoint(ShardId shardId, String codec) {
         length = 0L;
         this.codec = codec;
         this.metadataMap = Collections.emptyMap();
-        this.createdTimeStamp = System.currentTimeMillis();
+        this.createdTimeStamp = System.nanoTime();
     }
 
     public ReplicationCheckpoint(ShardId shardId, long primaryTerm, long segmentsGen, long segmentInfosVersion, String codec) {
-        this(shardId, primaryTerm, segmentsGen, segmentInfosVersion, 0L, codec, Collections.emptyMap(), System.currentTimeMillis());
+        this(shardId, primaryTerm, segmentsGen, segmentInfosVersion, 0L, codec, Collections.emptyMap(), System.nanoTime());
     }
 
     public ReplicationCheckpoint(

diff --git a/server/src/test/java/org/opensearch/indices/replication/SegmentReplicatorTests.java b/server/src/test/java/org/opensearch/indices/replication/SegmentReplicatorTests.java
@@ -234,7 +234,7 @@ public void testGetSegmentReplicationStats_WhenOnGoingReplication() {
             1000,
             "",
             stringStoreFileMetadataMap,
-            System.currentTimeMillis()
+            System.nanoTime()
         );
 
         segmentReplicator.updateReplicationCheckpoints(secondReplicationCheckpoint, shardId);
@@ -265,7 +265,7 @@ public void testGetSegmentReplicationStats_WhileOnGoingReplicationPrimaryRefresh
             1000,
             "",
             stringStoreFileMetadataMap,
-            System.currentTimeMillis()
+            System.nanoTime()
         );
 
         StoreFileMetadata storeFileMetadata3 = new StoreFileMetadata("test-3", 200, "1", Version.LATEST, bytesRef);
@@ -279,7 +279,7 @@ public void testGetSegmentReplicationStats_WhileOnGoingReplicationPrimaryRefresh
             1200,
             "",
             stringStoreFileMetadataMap,
-            System.currentTimeMillis()
+            System.nanoTime()
         );
 
         segmentReplicator.updateReplicationCheckpoints(secondReplicationCheckpoint, shardId);