From accf851a93d2ffa49eb9e28fa4fcb1f75af4bed3 Mon Sep 17 00:00:00 2001 From: Ashish Date: Sat, 9 Mar 2024 07:58:27 +0530 Subject: [PATCH] [Remote Store] Add check for byte lag in time lag calculation (#12565) Signed-off-by: Ashish Singh --- .../opensearch/index/remote/RemoteSegmentTransferTracker.java | 2 +- .../index/remote/RemoteSegmentTransferTrackerTests.java | 2 ++ .../index/remote/RemoteStorePressureServiceTests.java | 2 ++ 3 files changed, 5 insertions(+), 1 deletion(-) diff --git a/server/src/main/java/org/opensearch/index/remote/RemoteSegmentTransferTracker.java b/server/src/main/java/org/opensearch/index/remote/RemoteSegmentTransferTracker.java index 92436a09a4e7e..f1843ea3eef38 100644 --- a/server/src/main/java/org/opensearch/index/remote/RemoteSegmentTransferTracker.java +++ b/server/src/main/java/org/opensearch/index/remote/RemoteSegmentTransferTracker.java @@ -260,7 +260,7 @@ public long getRefreshSeqNoLag() { } public long getTimeMsLag() { - if (remoteRefreshTimeMs == localRefreshTimeMs) { + if (remoteRefreshTimeMs == localRefreshTimeMs || bytesLag == 0) { return 0; } return currentTimeMsUsingSystemNanos() - remoteRefreshStartTimeMs; diff --git a/server/src/test/java/org/opensearch/index/remote/RemoteSegmentTransferTrackerTests.java b/server/src/test/java/org/opensearch/index/remote/RemoteSegmentTransferTrackerTests.java index c87cdfcc8f1a1..ccdd1fe4ab609 100644 --- a/server/src/test/java/org/opensearch/index/remote/RemoteSegmentTransferTrackerTests.java +++ b/server/src/test/java/org/opensearch/index/remote/RemoteSegmentTransferTrackerTests.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.util.HashMap; +import java.util.List; import java.util.Map; import static org.opensearch.index.remote.RemoteSegmentTransferTracker.currentTimeMsUsingSystemNanos; @@ -149,6 +150,7 @@ public void testComputeTimeLagOnUpdate() throws InterruptedException { Thread.sleep(1); transferTracker.updateLocalRefreshTimeMs(currentTimeMsUsingSystemNanos()); + transferTracker.updateLatestLocalFileNameLengthMap(List.of("test"), k -> 1L); // Sleep for 100ms and then the lag should be within 100ms +/- 20ms Thread.sleep(100); assertTrue(Math.abs(transferTracker.getTimeMsLag() - 100) <= 20); diff --git a/server/src/test/java/org/opensearch/index/remote/RemoteStorePressureServiceTests.java b/server/src/test/java/org/opensearch/index/remote/RemoteStorePressureServiceTests.java index cb77174e612fd..9d00cf9f2be46 100644 --- a/server/src/test/java/org/opensearch/index/remote/RemoteStorePressureServiceTests.java +++ b/server/src/test/java/org/opensearch/index/remote/RemoteStorePressureServiceTests.java @@ -19,6 +19,7 @@ import org.opensearch.threadpool.ThreadPool; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicLong; import java.util.regex.Matcher; @@ -100,6 +101,7 @@ public void testValidateSegmentUploadLag() throws InterruptedException { while (currentTimeMsUsingSystemNanos() - localRefreshTimeMs <= 20 * avg) { Thread.sleep((long) (4 * avg)); } + pressureTracker.updateLatestLocalFileNameLengthMap(List.of("test"), k -> 1L); Exception e = assertThrows(OpenSearchRejectedExecutionException.class, () -> pressureService.validateSegmentsUploadLag(shardId)); String regex = "^rejected execution on primary shard:\\[index]\\[0] due to remote segments lagging behind " + "local segments.time_lag:[0-9]{2,3} ms dynamic_time_lag_threshold:95\\.0 ms$";