Add jitter for remote download calls

Signed-off-by: Arpit Bandejiya <abandeji@amazon.com>
opensearch-project · Sep 20, 2024 · 2bed915 · 2bed915
1 parent b2a7136
commit 2bed915
Show file tree

Hide file tree

Showing 38 changed files with 385 additions and 193 deletions.
diff --git a/...usterTest/java/org/opensearch/cluster/coordination/UnsafeBootstrapAndDetachCommandIT.java b/...usterTest/java/org/opensearch/cluster/coordination/UnsafeBootstrapAndDetachCommandIT.java
@@ -44,7 +44,7 @@
 import org.opensearch.env.TestEnvironment;
 import org.opensearch.gateway.GatewayMetaState;
 import org.opensearch.gateway.PersistedClusterStateService;
-import org.opensearch.gateway.remote.RemoteClusterStateService;
+import org.opensearch.gateway.remote.RemoteClusterStateSettings;
 import org.opensearch.indices.IndicesService;
 import org.opensearch.node.Node.DiscoverySettings;
 import org.opensearch.test.InternalTestCluster;
@@ -185,7 +185,7 @@ public void testBootstrapRemoteClusterEnabled() {
  final Environment environment = TestEnvironment.newEnvironment(
  Settings.builder()
  .put(internalCluster().getDefaultSettings())
- .put(RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING.getKey(), true)
+ .put(RemoteClusterStateSettings.REMOTE_CLUSTER_STATE_ENABLED_SETTING.getKey(), true)
  .build()
  );
  expectThrows(() -> unsafeBootstrap(environment), UnsafeBootstrapClusterManagerCommand.REMOTE_CLUSTER_STATE_ENABLED_NODE);

diff --git a/...nalClusterTest/java/org/opensearch/gateway/remote/RemoteClusterStateCleanupManagerIT.java b/...nalClusterTest/java/org/opensearch/gateway/remote/RemoteClusterStateCleanupManagerIT.java
@@ -39,8 +39,8 @@
 import static org.opensearch.gateway.remote.RemoteClusterStateCleanupManager.REMOTE_CLUSTER_STATE_CLEANUP_INTERVAL_SETTING;
 import static org.opensearch.gateway.remote.RemoteClusterStateCleanupManager.RETAINED_MANIFESTS;
 import static org.opensearch.gateway.remote.RemoteClusterStateCleanupManager.SKIP_CLEANUP_STATE_CHANGES;
-import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
-import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_PUBLICATION_SETTING_KEY;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_PUBLICATION_SETTING_KEY;
 import static org.opensearch.gateway.remote.RemoteUploadStats.REMOTE_UPLOAD;
 import static org.opensearch.gateway.remote.routingtable.RemoteIndexRoutingTable.INDEX_ROUTING_TABLE;
 import static org.opensearch.indices.IndicesService.CLUSTER_DEFAULT_INDEX_REFRESH_INTERVAL_SETTING;

diff --git a/...c/internalClusterTest/java/org/opensearch/gateway/remote/RemoteClusterStateServiceIT.java b/...c/internalClusterTest/java/org/opensearch/gateway/remote/RemoteClusterStateServiceIT.java
@@ -26,7 +26,7 @@
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
-import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
 import static org.opensearch.gateway.remote.RemoteClusterStateUtils.DELIMITER;
 import static org.opensearch.gateway.remote.RemoteClusterStateUtils.METADATA_FILE_PREFIX;
 import static org.opensearch.gateway.remote.model.RemoteCoordinationMetadata.COORDINATION_METADATA;

diff --git a/...ternalClusterTest/java/org/opensearch/gateway/remote/RemoteClusterStateTermVersionIT.java b/...ternalClusterTest/java/org/opensearch/gateway/remote/RemoteClusterStateTermVersionIT.java
@@ -39,8 +39,8 @@
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
-import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_PUBLICATION_SETTING_KEY;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_PUBLICATION_SETTING_KEY;
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY;
 import static org.hamcrest.Matchers.is;
 

diff --git a/...c/internalClusterTest/java/org/opensearch/gateway/remote/RemoteRoutingTableServiceIT.java b/...c/internalClusterTest/java/org/opensearch/gateway/remote/RemoteRoutingTableServiceIT.java
@@ -37,8 +37,9 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
-import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_PUBLICATION_SETTING_KEY;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_CLUSTER_STATE_CHECKSUM_VALIDATION_MODE_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_PUBLICATION_SETTING_KEY;
 import static org.opensearch.gateway.remote.routingtable.RemoteIndexRoutingTable.INDEX_ROUTING_TABLE;
 import static org.opensearch.indices.IndicesService.CLUSTER_DEFAULT_INDEX_REFRESH_INTERVAL_SETTING;
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY;
@@ -68,8 +69,8 @@ protected Settings nodeSettings(int nodeOrdinal) {
  .put("node.attr." + REMOTE_STORE_ROUTING_TABLE_REPOSITORY_NAME_ATTRIBUTE_KEY, REMOTE_ROUTING_TABLE_REPO)
  .put(REMOTE_PUBLICATION_SETTING_KEY, true)
  .put(
- RemoteClusterStateService.REMOTE_CLUSTER_STATE_CHECKSUM_VALIDATION_MODE_SETTING.getKey(),
- RemoteClusterStateService.RemoteClusterStateValidationMode.FAILURE
+ REMOTE_CLUSTER_STATE_CHECKSUM_VALIDATION_MODE_SETTING.getKey(),
+ RemoteClusterStateSettings.RemoteClusterStateValidationMode.FAILURE
  )
  .build();
  }

diff --git a/.../src/internalClusterTest/java/org/opensearch/gateway/remote/RemoteStatePublicationIT.java b/.../src/internalClusterTest/java/org/opensearch/gateway/remote/RemoteStatePublicationIT.java
@@ -52,9 +52,11 @@
 import static org.opensearch.action.admin.cluster.node.stats.NodesStatsRequest.Metric.DISCOVERY;
 import static org.opensearch.cluster.metadata.Metadata.isGlobalStateEquals;
 import static org.opensearch.gateway.remote.RemoteClusterStateAttributesManager.DISCOVERY_NODES;
-import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
-import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_PUBLICATION_SETTING;
-import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_PUBLICATION_SETTING_KEY;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.CLUSTER_REMOTE_STORE_STATE_PATH_PREFIX;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_CLUSTER_STATE_CHECKSUM_VALIDATION_MODE_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_PUBLICATION_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_PUBLICATION_SETTING_KEY;
 import static org.opensearch.gateway.remote.RemoteClusterStateUtils.DELIMITER;
 import static org.opensearch.gateway.remote.RemoteDownloadStats.CHECKSUM_VALIDATION_FAILED_COUNT;
 import static org.opensearch.gateway.remote.model.RemoteClusterBlocks.CLUSTER_BLOCKS;
@@ -112,14 +114,11 @@ protected Settings nodeSettings(int nodeOrdinal) {
  .put(routingTableRepoTypeAttributeKey, ReloadableFsRepository.TYPE)
  .put(routingTableRepoSettingsAttributeKeyPrefix + "location", segmentRepoPath)
  .put(
- RemoteClusterStateService.REMOTE_CLUSTER_STATE_CHECKSUM_VALIDATION_MODE_SETTING.getKey(),
- RemoteClusterStateService.RemoteClusterStateValidationMode.FAILURE
+ REMOTE_CLUSTER_STATE_CHECKSUM_VALIDATION_MODE_SETTING.getKey(),
+ RemoteClusterStateSettings.RemoteClusterStateValidationMode.FAILURE
  )
  .put(REMOTE_PUBLICATION_SETTING_KEY, isRemotePublicationEnabled)
- .put(
- RemoteClusterStateService.CLUSTER_REMOTE_STORE_STATE_PATH_PREFIX.getKey(),
- hasRemoteStateCharPrefix ? REMOTE_STATE_PREFIX : ""
- )
+ .put(CLUSTER_REMOTE_STORE_STATE_PATH_PREFIX.getKey(), hasRemoteStateCharPrefix ? REMOTE_STATE_PREFIX : "")
  .put(
  RemoteRoutingTableBlobStore.CLUSTER_REMOTE_STORE_ROUTING_TABLE_PATH_PREFIX.getKey(),
  hasRemoteRoutingCharPrefix ? REMOTE_ROUTING_PREFIX : ""

diff --git a/...er/src/internalClusterTest/java/org/opensearch/remotemigration/MigrationBaseTestCase.java b/...er/src/internalClusterTest/java/org/opensearch/remotemigration/MigrationBaseTestCase.java
@@ -48,7 +48,7 @@
 
 import static org.opensearch.cluster.metadata.IndexMetadata.SETTING_NUMBER_OF_REPLICAS;
 import static org.opensearch.cluster.routing.allocation.decider.EnableAllocationDecider.CLUSTER_ROUTING_REBALANCE_ENABLE_SETTING;
-import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
 import static org.opensearch.node.remotestore.RemoteStoreNodeService.MIGRATION_DIRECTION_SETTING;
 import static org.opensearch.node.remotestore.RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING;
 import static org.opensearch.repositories.fs.ReloadableFsRepository.REPOSITORIES_FAILRATE_SETTING;

diff --git a/...internalClusterTest/java/org/opensearch/remotestore/RemoteStoreClusterStateRestoreIT.java b/...internalClusterTest/java/org/opensearch/remotestore/RemoteStoreClusterStateRestoreIT.java
@@ -58,7 +58,7 @@
 import static org.opensearch.cluster.metadata.IndexMetadata.INDEX_READ_ONLY_SETTING;
 import static org.opensearch.cluster.metadata.Metadata.CLUSTER_READ_ONLY_BLOCK;
 import static org.opensearch.cluster.metadata.Metadata.SETTING_READ_ONLY_SETTING;
-import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
 import static org.opensearch.gateway.remote.RemoteClusterStateUtils.encodeString;
 import static org.opensearch.indices.ShardLimitValidator.SETTING_CLUSTER_MAX_SHARDS_PER_NODE;
 import static org.opensearch.repositories.blobstore.BlobStoreRepository.SYSTEM_REPOSITORY_SETTING;

diff --git a/...src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreUploadIndexPathIT.java b/...src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreUploadIndexPathIT.java
@@ -22,7 +22,7 @@
 import java.util.Arrays;
 import java.util.concurrent.ExecutionException;
 
-import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
 import static org.opensearch.indices.RemoteStoreSettings.CLUSTER_REMOTE_STORE_PATH_TYPE_SETTING;
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
 

diff --git a/server/src/main/java/org/opensearch/cluster/coordination/CoordinationState.java b/server/src/main/java/org/opensearch/cluster/coordination/CoordinationState.java
@@ -53,7 +53,7 @@
 import java.util.Set;
 
 import static org.opensearch.cluster.coordination.Coordinator.ZEN1_BWC_TERM;
-import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_PUBLICATION_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_PUBLICATION_SETTING;
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.isRemoteStoreClusterStateEnabled;
 
 /**

diff --git a/server/src/main/java/org/opensearch/cluster/coordination/ElectionSchedulerFactory.java b/server/src/main/java/org/opensearch/cluster/coordination/ElectionSchedulerFactory.java
@@ -156,7 +156,7 @@ public Releasable startElectionScheduler(TimeValue gracePeriod, Runnable schedul
  }
 
  @SuppressForbidden(reason = "Argument to Math.abs() is definitely not Long.MIN_VALUE")
- private static long nonNegative(long n) {
+ public static long nonNegative(long n) {
  return n == Long.MIN_VALUE ? 0 : Math.abs(n);
  }
 

diff --git a/...c/main/java/org/opensearch/cluster/coordination/UnsafeBootstrapClusterManagerCommand.java b/...c/main/java/org/opensearch/cluster/coordination/UnsafeBootstrapClusterManagerCommand.java
@@ -53,7 +53,7 @@
 import java.util.Locale;
 import java.util.Objects;
 
-import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
 
 /**
  * Tool to run an unsafe bootstrap

diff --git a/...rc/main/java/org/opensearch/cluster/routing/remote/InternalRemoteRoutingTableService.java b/...rc/main/java/org/opensearch/cluster/routing/remote/InternalRemoteRoutingTableService.java
@@ -27,6 +27,8 @@
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.compress.Compressor;
 import org.opensearch.gateway.remote.ClusterMetadataManifest;
+import org.opensearch.gateway.remote.DefaultRandomObject;
+import org.opensearch.gateway.remote.RemoteClusterStateSettings;
 import org.opensearch.gateway.remote.RemoteClusterStateUtils;
 import org.opensearch.gateway.remote.RemoteStateTransferException;
 import org.opensearch.gateway.remote.model.RemoteRoutingTableBlobStore;
@@ -45,10 +47,12 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Random;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
+import static org.opensearch.gateway.remote.RemoteClusterStateUtils.toPositiveLongAtMost;
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.isRemoteRoutingTableEnabled;
 
 /**
@@ -68,11 +72,14 @@ public class InternalRemoteRoutingTableService extends AbstractLifecycleComponen
  private BlobStoreRepository blobStoreRepository;
  private final ThreadPool threadPool;
  private final String clusterName;
+ private RemoteClusterStateSettings remoteClusterStateSettings;
+ private Random random;
 
  public InternalRemoteRoutingTableService(
  Supplier<RepositoriesService> repositoriesService,
  Settings settings,
  ClusterSettings clusterSettings,
+ RemoteClusterStateSettings remoteClusterStateSettings,
  ThreadPool threadpool,
  String clusterName
  ) {
@@ -82,6 +89,8 @@ public InternalRemoteRoutingTableService(
  this.threadPool = threadpool;
  this.clusterName = clusterName;
  this.clusterSettings = clusterSettings;
+ this.remoteClusterStateSettings = remoteClusterStateSettings;
+ this.random = DefaultRandomObject.INSTANCE;
  }
 
  public List<IndexRoutingTable> getIndicesRouting(RoutingTable routingTable) {
@@ -193,7 +202,10 @@ public void getAsyncIndexRoutingReadAction(
 
  RemoteIndexRoutingTable remoteIndexRoutingTable = new RemoteIndexRoutingTable(uploadedFilename, clusterUUID, compressor);
 
- remoteIndexRoutingTableStore.readAsync(remoteIndexRoutingTable, actionListener);
+ long maxDelayInMillis = remoteClusterStateSettings.getRemoteStateReadMaxJitter().getMillis();
+ final long delayInMillis = toPositiveLongAtMost(random.nextLong(), maxDelayInMillis);
+ remoteIndexRoutingTableStore.readAsyncWithDelay(delayInMillis, remoteIndexRoutingTable, actionListener);
+
  }
 
  @Override
@@ -208,7 +220,10 @@ public void getAsyncIndexRoutingTableDiffReadAction(
  );
 
  RemoteRoutingTableDiff remoteRoutingTableDiff = new RemoteRoutingTableDiff(uploadedFilename, clusterUUID, compressor);
- remoteRoutingTableDiffStore.readAsync(remoteRoutingTableDiff, actionListener);
+
+ long maxDelayInMillis = remoteClusterStateSettings.getRemoteStateReadMaxJitter().getMillis();
+ final long delayInMillis = toPositiveLongAtMost(random.nextLong(), maxDelayInMillis);
+ remoteRoutingTableDiffStore.readAsyncWithDelay(delayInMillis, remoteRoutingTableDiff, actionListener);
  }
 
  @Override

diff --git a/...src/main/java/org/opensearch/cluster/routing/remote/RemoteRoutingTableServiceFactory.java b/...src/main/java/org/opensearch/cluster/routing/remote/RemoteRoutingTableServiceFactory.java
@@ -10,6 +10,7 @@
 
 import org.opensearch.common.settings.ClusterSettings;
 import org.opensearch.common.settings.Settings;
+import org.opensearch.gateway.remote.RemoteClusterStateSettings;
 import org.opensearch.repositories.RepositoriesService;
 import org.opensearch.threadpool.ThreadPool;
 
@@ -34,11 +35,19 @@ public static RemoteRoutingTableService getService(
  Supplier<RepositoriesService> repositoriesService,
  Settings settings,
  ClusterSettings clusterSettings,
+ RemoteClusterStateSettings remoteClusterStateSettings,
  ThreadPool threadPool,
  String clusterName
  ) {
  if (isRemoteRoutingTableEnabled(settings)) {
- return new InternalRemoteRoutingTableService(repositoriesService, settings, clusterSettings, threadPool, clusterName);
+ return new InternalRemoteRoutingTableService(
+ repositoriesService,
+ settings,
+ clusterSettings,
+ remoteClusterStateSettings,
+ threadPool,
+ clusterName
+ );
  }
  return new NoopRemoteRoutingTableService();
  }

diff --git a/server/src/main/java/org/opensearch/common/remote/AbstractRemoteWritableEntityManager.java b/server/src/main/java/org/opensearch/common/remote/AbstractRemoteWritableEntityManager.java
@@ -8,12 +8,19 @@
 
 package org.opensearch.common.remote;
 
+import org.opensearch.common.settings.ClusterSettings;
+import org.opensearch.common.settings.Settings;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.gateway.remote.ClusterMetadataManifest;
+import org.opensearch.gateway.remote.DefaultRandomObject;
+import org.opensearch.gateway.remote.RemoteClusterStateSettings;
 import org.opensearch.gateway.remote.model.RemoteReadResult;
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Random;
+
+import static org.opensearch.gateway.remote.RemoteClusterStateUtils.toPositiveLongAtMost;
 
 /**
  * An abstract class that provides a base implementation for managing remote entities in the remote store.
@@ -24,6 +31,13 @@ public abstract class AbstractRemoteWritableEntityManager implements RemoteWrita
  */
  protected final Map<String, RemoteWritableEntityStore> remoteWritableEntityStores = new HashMap<>();
 
+ protected RemoteClusterStateSettings remoteClusterStateSettings = new RemoteClusterStateSettings(
+ Settings.EMPTY,
+ new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS)
+ );
+
+ protected final Random random = DefaultRandomObject.INSTANCE;
+
  /**
  * Retrieves the remote writable entity store for the given entity.
  *
@@ -79,6 +93,9 @@ public void writeAsync(
 
  @Override
  public void readAsync(String component, AbstractClusterMetadataWriteableBlobEntity entity, ActionListener<RemoteReadResult> listener) {
- getStore(entity).readAsync(entity, getWrappedReadListener(component, entity, listener));
+ long maxDelayInMillis = remoteClusterStateSettings.getRemoteStateReadMaxJitter().getMillis();
+ final long delayInMillis = toPositiveLongAtMost(random.nextLong(), maxDelayInMillis);
+ getStore(entity).readAsyncWithDelay(delayInMillis, entity, getWrappedReadListener(component, entity, listener));
  }
+
 }
diff --git a/server/src/main/java/org/opensearch/common/remote/RemoteWritableEntityStore.java b/server/src/main/java/org/opensearch/common/remote/RemoteWritableEntityStore.java
@@ -27,4 +27,6 @@ public interface RemoteWritableEntityStore<T, U extends RemoteWriteableEntity<T>
  public T read(U entity) throws IOException;
 
  public void readAsync(U entity, ActionListener<T> listener);
+
+ public void readAsyncWithDelay(long delayInMillis, U entity, ActionListener<T> listener);
 }
diff --git a/server/src/main/java/org/opensearch/common/remote/RemoteWriteableEntityBlobStore.java b/server/src/main/java/org/opensearch/common/remote/RemoteWriteableEntityBlobStore.java
@@ -10,6 +10,7 @@
 
 import org.opensearch.common.blobstore.BlobPath;
 import org.opensearch.common.blobstore.stream.write.WritePriority;
+import org.opensearch.common.unit.TimeValue;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.index.translog.transfer.BlobStoreTransferService;
 import org.opensearch.repositories.blobstore.BlobStoreRepository;
@@ -32,8 +33,10 @@ public class RemoteWriteableEntityBlobStore<T, U extends RemoteWriteableBlobEnti
  private final BlobStoreTransferService transferService;
  private final BlobStoreRepository blobStoreRepository;
  private final String clusterName;
+ private final String executor;
  private final ExecutorService executorService;
  private final String pathToken;
+ private final ThreadPool threadPool;
 
  public RemoteWriteableEntityBlobStore(
  final BlobStoreTransferService blobStoreTransferService,
@@ -46,6 +49,8 @@ public RemoteWriteableEntityBlobStore(
  this.transferService = blobStoreTransferService;
  this.blobStoreRepository = blobStoreRepository;
  this.clusterName = clusterName;
+ this.threadPool = threadPool;
+ this.executor = executor;
  this.executorService = threadPool.executor(executor);
  this.pathToken = pathToken;
  }
@@ -89,6 +94,17 @@ public void readAsync(final U entity, final ActionListener<T> listener) {
  });
  }
 
+ public void readAsyncWithDelay(final long delayMillis, final U entity, final ActionListener<T> listener) {
+ Runnable runnable = () -> {
+ try {
+ listener.onResponse(read(entity));
+ } catch (Exception e) {
+ listener.onFailure(e);
+ }
+ };
+ threadPool.scheduleUnlessShuttingDown(TimeValue.timeValueMillis(delayMillis), executor, runnable);
+ }
+
  public String getClusterName() {
  return clusterName;
  }
@@ -121,5 +137,4 @@ public BlobPath getBlobPathForDownload(final RemoteWriteableBlobEntity<T> obj) {
  private static String encodeString(String content) {
  return Base64.getUrlEncoder().withoutPadding().encodeToString(content.getBytes(StandardCharsets.UTF_8));
  }
-
 }