[Remote Store] Add segment transfer timeout dynamic setting

server/src/main/java/org/opensearch/common/settings/ClusterSettings.java

@@ -737,6 +737,7 @@ public void apply(Settings value, Settings current, Settings previous) {
                 RemoteStoreSettings.CLUSTER_REMOTE_INDEX_SEGMENT_METADATA_RETENTION_MAX_COUNT_SETTING,
                 RemoteStoreSettings.CLUSTER_REMOTE_TRANSLOG_BUFFER_INTERVAL_SETTING,
                 RemoteStoreSettings.CLUSTER_REMOTE_TRANSLOG_TRANSFER_TIMEOUT_SETTING,
+                RemoteStoreSettings.CLUSTER_REMOTE_SEGMENT_TRANSFER_TIMEOUT_SETTING,
                 RemoteStoreSettings.CLUSTER_REMOTE_STORE_PATH_TYPE_SETTING,
                 RemoteStoreSettings.CLUSTER_REMOTE_STORE_PATH_HASH_ALGORITHM_SETTING,
                 RemoteStoreSettings.CLUSTER_REMOTE_MAX_TRANSLOG_READERS

server/src/main/java/org/opensearch/index/shard/IndexShard.java

@@ -3970,7 +3970,8 @@ private EngineConfig newEngineConfig(LongSupplier globalCheckpointSupplier) thro
                 new RemoteStoreRefreshListener(
                     this,
                     this.checkpointPublisher,
-                    remoteStoreStatsTrackerFactory.getRemoteSegmentTransferTracker(shardId())
+                    remoteStoreStatsTrackerFactory.getRemoteSegmentTransferTracker(shardId()),
+                    remoteStoreSettings
                 )
             );
         }

server/src/main/java/org/opensearch/index/shard/RemoteStoreRefreshListener.java

@@ -33,6 +33,7 @@
 import org.opensearch.index.store.RemoteSegmentStoreDirectory;
 import org.opensearch.index.store.remote.metadata.RemoteSegmentMetadata;
 import org.opensearch.index.translog.Translog;
+import org.opensearch.indices.RemoteStoreSettings;
 import org.opensearch.indices.replication.checkpoint.ReplicationCheckpoint;
 import org.opensearch.indices.replication.checkpoint.SegmentReplicationCheckpointPublisher;
 import org.opensearch.threadpool.ThreadPool;
@@ -45,6 +46,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
@@ -89,11 +91,13 @@ public final class RemoteStoreRefreshListener extends ReleasableRetryableRefresh
     private volatile long primaryTerm;
     private volatile Iterator<TimeValue> backoffDelayIterator;
     private final SegmentReplicationCheckpointPublisher checkpointPublisher;
+    private final RemoteStoreSettings remoteStoreSettings;
 
     public RemoteStoreRefreshListener(
         IndexShard indexShard,
         SegmentReplicationCheckpointPublisher checkpointPublisher,
-        RemoteSegmentTransferTracker segmentTracker
+        RemoteSegmentTransferTracker segmentTracker,
+        RemoteStoreSettings remoteStoreSettings
     ) {
         super(indexShard.getThreadPool());
         logger = Loggers.getLogger(getClass(), indexShard.shardId());
@@ -116,6 +120,7 @@ public RemoteStoreRefreshListener(
         this.segmentTracker = segmentTracker;
         resetBackOffDelayIterator();
         this.checkpointPublisher = checkpointPublisher;
+        this.remoteStoreSettings = remoteStoreSettings;
     }
 
     @Override
@@ -286,7 +291,12 @@ public void onFailure(Exception e) {
 
                     // Start the segments files upload
                     uploadNewSegments(localSegmentsPostRefresh, localSegmentsSizeMap, segmentUploadsCompletedListener);
-                    latch.await();
+                    if (latch.await(
+                        remoteStoreSettings.getClusterRemoteSegmentTransferTimeout().millis(),
+                        TimeUnit.MILLISECONDS
+                    ) == false) {
+                        throw new InterruptedException("Timeout while waiting for remote segment transfer to complete");
+                    }
                 } catch (EngineException e) {
                     logger.warn("Exception while reading SegmentInfosSnapshot", e);
                 }

server/src/main/java/org/opensearch/indices/RemoteStoreSettings.java

@@ -105,9 +105,21 @@ public class RemoteStoreSettings {
         Property.NodeScope
     );
 
+    /**
+     * Controls timeout value while uploading segment files to remote segment store
+     */
+    public static final Setting<TimeValue> CLUSTER_REMOTE_SEGMENT_TRANSFER_TIMEOUT_SETTING = Setting.timeSetting(
+        "cluster.remote_store.segment.transfer_timeout",
+        TimeValue.timeValueHours(3),
+        TimeValue.timeValueMinutes(10),
+        Property.NodeScope,
+        Property.Dynamic
+    );
+
     private volatile TimeValue clusterRemoteTranslogBufferInterval;
     private volatile int minRemoteSegmentMetadataFiles;
     private volatile TimeValue clusterRemoteTranslogTransferTimeout;
+    private volatile TimeValue clusterRemoteSegmentTransferTimeout;
     private volatile RemoteStoreEnums.PathType pathType;
     private volatile RemoteStoreEnums.PathHashAlgorithm pathHashAlgorithm;
     private volatile int maxRemoteTranslogReaders;
@@ -139,6 +151,12 @@ public RemoteStoreSettings(Settings settings, ClusterSettings clusterSettings) {
 
         maxRemoteTranslogReaders = CLUSTER_REMOTE_MAX_TRANSLOG_READERS.get(settings);
         clusterSettings.addSettingsUpdateConsumer(CLUSTER_REMOTE_MAX_TRANSLOG_READERS, this::setMaxRemoteTranslogReaders);
+
+        clusterRemoteSegmentTransferTimeout = CLUSTER_REMOTE_SEGMENT_TRANSFER_TIMEOUT_SETTING.get(settings);
+        clusterSettings.addSettingsUpdateConsumer(
+            CLUSTER_REMOTE_SEGMENT_TRANSFER_TIMEOUT_SETTING,
+            this::setClusterRemoteSegmentTransferTimeout
+        );
     }
 
     public TimeValue getClusterRemoteTranslogBufferInterval() {
@@ -161,10 +179,18 @@ public TimeValue getClusterRemoteTranslogTransferTimeout() {
         return clusterRemoteTranslogTransferTimeout;
     }
 
+    public TimeValue getClusterRemoteSegmentTransferTimeout() {
+        return clusterRemoteSegmentTransferTimeout;
+    }
+
     private void setClusterRemoteTranslogTransferTimeout(TimeValue clusterRemoteTranslogTransferTimeout) {
         this.clusterRemoteTranslogTransferTimeout = clusterRemoteTranslogTransferTimeout;
     }
 
+    private void setClusterRemoteSegmentTransferTimeout(TimeValue clusterRemoteSegmentTransferTimeout) {
+        this.clusterRemoteSegmentTransferTimeout = clusterRemoteSegmentTransferTimeout;
+    }
+
     @ExperimentalApi
     public RemoteStoreEnums.PathType getPathType() {
         return pathType;

server/src/test/java/org/opensearch/index/shard/RemoteStoreRefreshListenerTests.java

@@ -34,6 +34,7 @@
 import org.opensearch.index.store.RemoteSegmentStoreDirectory.MetadataFilenameUtils;
 import org.opensearch.index.store.Store;
 import org.opensearch.index.store.lockmanager.RemoteStoreLockManager;
+import org.opensearch.indices.DefaultRemoteStoreSettings;
 import org.opensearch.indices.RemoteStoreSettings;
 import org.opensearch.indices.replication.checkpoint.SegmentReplicationCheckpointPublisher;
 import org.opensearch.indices.replication.common.ReplicationType;
@@ -90,7 +91,12 @@ public void setup(boolean primary, int numberOfDocs) throws IOException {
         remoteStoreStatsTrackerFactory = new RemoteStoreStatsTrackerFactory(clusterService, Settings.EMPTY);
         remoteStoreStatsTrackerFactory.afterIndexShardCreated(indexShard);
         RemoteSegmentTransferTracker tracker = remoteStoreStatsTrackerFactory.getRemoteSegmentTransferTracker(indexShard.shardId());
-        remoteStoreRefreshListener = new RemoteStoreRefreshListener(indexShard, SegmentReplicationCheckpointPublisher.EMPTY, tracker);
+        remoteStoreRefreshListener = new RemoteStoreRefreshListener(
+            indexShard,
+            SegmentReplicationCheckpointPublisher.EMPTY,
+            tracker,
+            DefaultRemoteStoreSettings.INSTANCE
+        );
     }
 
     private void indexDocs(int startDocId, int numberOfDocs) throws IOException {
@@ -175,7 +181,12 @@ public void testRemoteDirectoryInitThrowsException() throws IOException {
         when(remoteStore.directory()).thenReturn(remoteStoreFilterDirectory);
 
         // Since the thrown IOException is caught in the constructor, ctor should be invoked successfully.
-        new RemoteStoreRefreshListener(shard, SegmentReplicationCheckpointPublisher.EMPTY, mock(RemoteSegmentTransferTracker.class));
+        new RemoteStoreRefreshListener(
+            shard,
+            SegmentReplicationCheckpointPublisher.EMPTY,
+            mock(RemoteSegmentTransferTracker.class),
+            DefaultRemoteStoreSettings.INSTANCE
+        );
 
         // Validate that the stream of metadata file of remoteMetadataDirectory has been opened only once and the
         // listFilesByPrefixInLexicographicOrder has been called twice.
@@ -638,7 +649,12 @@ private Tuple<RemoteStoreRefreshListener, RemoteStoreStatsTrackerFactory> mockIn
         RemoteStoreSettings remoteStoreSettings = mock(RemoteStoreSettings.class);
         when(remoteStoreSettings.getMinRemoteSegmentMetadataFiles()).thenReturn(10);
         when(shard.getRemoteStoreSettings()).thenReturn(remoteStoreSettings);
-        RemoteStoreRefreshListener refreshListener = new RemoteStoreRefreshListener(shard, emptyCheckpointPublisher, tracker);
+        RemoteStoreRefreshListener refreshListener = new RemoteStoreRefreshListener(
+            shard,
+            emptyCheckpointPublisher,
+            tracker,
+            DefaultRemoteStoreSettings.INSTANCE
+        );
         refreshListener.afterRefresh(true);
         return Tuple.tuple(refreshListener, remoteStoreStatsTrackerFactory);
     }