Fix timeout issue by calculating time needed to process getSegmentFiles.

Signed-off-by: Rishikesh1159 <rishireddy1159@gmail.com>

CHANGELOG.md

@@ -47,6 +47,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - Fix NoSuchFileExceptions with segment replication when computing primary metadata snapshots ([#4366](https://github.com/opensearch-project/OpenSearch/pull/4366))
 - [Segment Replication] Update flaky testOnNewCheckpointFromNewPrimaryCancelOngoingReplication unit test ([#4414](https://github.com/opensearch-project/OpenSearch/pull/4414))
 - Fixed the `_cat/shards/10_basic.yml` test cases fix.
+- Fix timeout issue by calculating time needed to process getSegmentFiles
 
 ### Security
 - CVE-2022-25857 org.yaml:snakeyaml DOS vulnerability ([#4341](https://github.com/opensearch-project/OpenSearch/pull/4341))
@@ -71,4 +72,4 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 
 
 [Unreleased]: https://github.com/opensearch-project/OpenSearch/compare/2.2.0...HEAD
-[2.x]: https://github.com/opensearch-project/OpenSearch/compare/2.2.0...2.x
+[2.x]: https://github.com/opensearch-project/OpenSearch/compare/2.2.0...2.x

server/src/main/java/org/opensearch/indices/replication/PrimaryShardReplicationSource.java

@@ -13,11 +13,13 @@
 import org.opensearch.action.ActionListener;
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.common.io.stream.Writeable;
+import org.opensearch.common.unit.TimeValue;
 import org.opensearch.index.store.Store;
 import org.opensearch.index.store.StoreFileMetadata;
 import org.opensearch.indices.recovery.RecoverySettings;
 import org.opensearch.indices.recovery.RetryableTransportClient;
 import org.opensearch.indices.replication.checkpoint.ReplicationCheckpoint;
+import org.opensearch.transport.TransportRequestOptions;
 import org.opensearch.transport.TransportService;
 
 import java.util.List;
@@ -78,14 +80,33 @@ public void getSegmentFiles(
     ) {
         final Writeable.Reader<GetSegmentFilesResponse> reader = GetSegmentFilesResponse::new;
         final ActionListener<GetSegmentFilesResponse> responseListener = ActionListener.map(listener, r -> r);
+        // Few of the below assumptions and calculations are added for experimental release of segment replication feature in 2.3
+        // version.These will be changed in next release.
+
+        // Storing the size of files to fetch in bytes.
+        long sizeOfSegmentFiles = 0;
+        for (int i = 0; i < filesToFetch.size(); i++) {
+            sizeOfSegmentFiles += filesToFetch.get(i).length();
+        }
+        // Making sure files size is in correct format to perform time calculation.
+        sizeOfSegmentFiles = Math.abs(sizeOfSegmentFiles);
+        // Maximum size of files to fetch (segment files), that can be processed in 1 minute for a m5.xlarge machine.
+        long baseSegmentFilesSize = 300000000;
+
+        long timeToGetSegmentFiles = 1;
+        // Formula for calculating time needed to process a replication event's files to fetch process
+        timeToGetSegmentFiles += sizeOfSegmentFiles / baseSegmentFilesSize;
         final GetSegmentFilesRequest request = new GetSegmentFilesRequest(
             replicationId,
             targetAllocationId,
             targetNode,
             filesToFetch,
             checkpoint
         );
-        transportClient.executeRetryableAction(GET_SEGMENT_FILES, request, responseListener, reader);
+        final TransportRequestOptions options = TransportRequestOptions.builder()
+            .withTimeout(TimeValue.timeValueMinutes(timeToGetSegmentFiles))
+            .build();
+        transportClient.executeRetryableAction(GET_SEGMENT_FILES, request, options, responseListener, reader);
     }
 
     @Override