Restored RemoteSegmentStoreDirectory PublicAPI + added changelog

Signed-off-by: kh3ra <[email protected]>

Author: kh3ra

CHANGELOG.md

@@ -45,6 +45,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Added approximation support for range queries with now in date field ([#18511](https://github.com/opensearch-project/OpenSearch/pull/18511))
 - Upgrade to protobufs 0.6.0 and clean up deprecated TermQueryProtoUtils code ([#18880](https://github.com/opensearch-project/OpenSearch/pull/18880))
 - Prevent shard initialization failure due to streaming consumer errors ([#18877](https://github.com/opensearch-project/OpenSearch/pull/18877))
+- Added the core process for warming merged segments in remote-store enabled domains ([#18683](https://github.com/opensearch-project/OpenSearch/pull/18683))
 
 ### Changed
 - Update Subject interface to use CheckedRunnable ([#18570](https://github.com/opensearch-project/OpenSearch/issues/18570))

server/src/internalClusterTest/java/org/opensearch/indices/replication/RemoteStoreMergedSegmentWarmerIT.java

@@ -156,6 +156,7 @@ public void testConcurrentMergeSegmentWarmerRemote() throws Exception {
         logger.info("Number of merge invocations: {}", numInvocations.get());
         assertTrue(latch.await(10, TimeUnit.SECONDS));
         assertTrue(executingThreads.size() > 1);
+        // Verify concurrent execution by checking that multiple unique threads handled merge operations
         assertTrue(numInvocations.get() > 1);
         mockTransportServiceNode1.clearAllRules();
         mockTransportServiceNode2.clearAllRules();

server/src/main/java/org/opensearch/index/store/RemoteSegmentStoreDirectory.java

@@ -23,7 +23,9 @@
 import org.apache.lucene.store.IndexInput;
 import org.apache.lucene.store.IndexOutput;
 import org.apache.lucene.util.Version;
+import org.opensearch.common.Nullable;
 import org.opensearch.common.UUIDs;
+import org.opensearch.common.annotation.InternalApi;
 import org.opensearch.common.annotation.PublicApi;
 import org.opensearch.common.collect.Tuple;
 import org.opensearch.common.io.VersionedCodecStreamWrapper;
@@ -131,13 +133,24 @@ public final class RemoteSegmentStoreDirectory extends FilterDirectory implement
 
     public static final int METADATA_FILES_TO_FETCH = 10;
 
+    public RemoteSegmentStoreDirectory(
+        RemoteDirectory remoteDataDirectory,
+        RemoteDirectory remoteMetadataDirectory,
+        RemoteStoreLockManager mdLockManager,
+        ThreadPool threadPool,
+        ShardId shardId
+    ) throws IOException {
+        this(remoteDataDirectory, remoteMetadataDirectory, mdLockManager, threadPool, shardId, null);
+    }
+
+    @InternalApi
     public RemoteSegmentStoreDirectory(
         RemoteDirectory remoteDataDirectory,
         RemoteDirectory remoteMetadataDirectory,
         RemoteStoreLockManager mdLockManager,
         ThreadPool threadPool,
         ShardId shardId,
-        Map<String, String> pendingDownloadMergedSegments
+        @Nullable Map<String, String> pendingDownloadMergedSegments
     ) throws IOException {
         super(remoteDataDirectory);
         this.remoteDataDirectory = remoteDataDirectory;
@@ -1126,7 +1139,7 @@ public void close() throws IOException {
      *
      * @param localToRemoteFilenames Map of local filenames to their corresponding remote filenames
      */
-    public void markPendingMergedSegmentsDownload(Map<String, String> localToRemoteFilenames) {
+    public void markMergedSegmentsPendingDownload(Map<String, String> localToRemoteFilenames) {
         pendingDownloadMergedSegments.putAll(localToRemoteFilenames);
     }
 
@@ -1135,7 +1148,7 @@ public void markPendingMergedSegmentsDownload(Map<String, String> localToRemoteF
      *
      * @param localFilenames Set of local filenames to remove from pending downloads
      */
-    public void unmarkPendingDownloadMergedSegments(Set<String> localFilenames) {
+    public void unmarkMergedSegmentsPendingDownload(Set<String> localFilenames) {
         localFilenames.forEach(pendingDownloadMergedSegments::remove);
     }
 
@@ -1146,6 +1159,6 @@ public void unmarkPendingDownloadMergedSegments(Set<String> localFilenames) {
      * @return true if segment is pending download, false otherwise
      */
     public boolean isMergedSegmentPendingDownload(String localFilename) {
-        return pendingDownloadMergedSegments.containsKey(localFilename);
+        return pendingDownloadMergedSegments != null && pendingDownloadMergedSegments.containsKey(localFilename);
     }
 }

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

@@ -737,7 +737,7 @@ public void onReplicationFailure(
 
     private void unmarkPendingDownloadMergedSegments(IndexShard shard, ReplicationCheckpoint receivedCheckpoint) {
         if (isRemoteStoreMergedSegmentCheckpoint(receivedCheckpoint)) {
-            shard.getRemoteDirectory().unmarkPendingDownloadMergedSegments(receivedCheckpoint.getMetadataMap().keySet());
+            shard.getRemoteDirectory().unmarkMergedSegmentsPendingDownload(receivedCheckpoint.getMetadataMap().keySet());
         }
     }
 

server/src/main/java/org/opensearch/indices/replication/checkpoint/RemoteStorePublishMergedSegmentAction.java

@@ -83,7 +83,7 @@ public RemoteStorePublishMergedSegmentAction(
     protected void doReplicaOperation(RemoteStorePublishMergedSegmentRequest shardRequest, IndexShard replica) {
         RemoteStoreMergedSegmentCheckpoint checkpoint = shardRequest.getMergedSegment();
         if (checkpoint.getShardId().equals(replica.shardId())) {
-            replica.getRemoteDirectory().markPendingMergedSegmentsDownload(checkpoint.getLocalToRemoteSegmentFilenameMap());
+            replica.getRemoteDirectory().markMergedSegmentsPendingDownload(checkpoint.getLocalToRemoteSegmentFilenameMap());
             replicationService.onNewMergedSegmentCheckpoint(checkpoint, replica);
         } else {
             logger.warn(

server/src/test/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryTests.java

@@ -1201,6 +1201,26 @@ public void testInitializeToSpecificTimestampMatchingMdFile() throws IOException
         assertTrue(uploadedSegments.containsKey("_0.cfs"));
     }
 
+    public void testMarkMergedSegmentPendingDownload() {
+        String localSegmentName1 = "_1.si";
+        String remoteSegmentName1 = "_1.si__uuid";
+        String localSegmentName2 = "_1.dvd";
+        String remoteSegmentName2 = "_1.dvd__uuid";
+        remoteSegmentStoreDirectory.markMergedSegmentsPendingDownload(
+            Map.of(localSegmentName1, remoteSegmentName1, localSegmentName2, remoteSegmentName2)
+        );
+
+        assertTrue(remoteSegmentStoreDirectory.isMergedSegmentPendingDownload(localSegmentName1));
+        assertTrue(remoteSegmentStoreDirectory.isMergedSegmentPendingDownload(localSegmentName2));
+        assertFalse(remoteSegmentStoreDirectory.isMergedSegmentPendingDownload("_3.si"));
+        assertEquals(remoteSegmentName1, remoteSegmentStoreDirectory.getExistingRemoteFilename(localSegmentName1));
+        assertEquals(remoteSegmentName2, remoteSegmentStoreDirectory.getExistingRemoteFilename(localSegmentName2));
+
+        remoteSegmentStoreDirectory.unmarkMergedSegmentsPendingDownload(Set.of(localSegmentName1));
+        assertFalse(remoteSegmentStoreDirectory.isMergedSegmentPendingDownload(localSegmentName1));
+        assertNull(remoteSegmentStoreDirectory.getExistingRemoteFilename(localSegmentName1));
+    }
+
     private static class WrapperIndexOutput extends IndexOutput {
         public IndexOutput indexOutput;
 

server/src/test/java/org/opensearch/indices/replication/RemoteStoreReplicationSourceTests.java

@@ -202,7 +202,7 @@ public void testGetMergedSegmentFiles() throws IOException, ExecutionException,
                 });
             }
         };
-        replicaShard.getRemoteDirectory().markPendingMergedSegmentsDownload(localToRemoteFilenameMap);
+        replicaShard.getRemoteDirectory().markMergedSegmentsPendingDownload(localToRemoteFilenameMap);
         RemoteStoreMergedSegmentCheckpoint mergedSegmentCheckpoint = new RemoteStoreMergedSegmentCheckpoint(
             new MergedSegmentCheckpoint(
                 replicaShard.shardId(),
@@ -243,7 +243,7 @@ public void testGetMergedSegmentFilesDownloadTimeout() throws IOException, Execu
                 });
             }
         };
-        replicaShard.getRemoteDirectory().markPendingMergedSegmentsDownload(localToRemoteFilenameMap);
+        replicaShard.getRemoteDirectory().markMergedSegmentsPendingDownload(localToRemoteFilenameMap);
         RemoteStoreMergedSegmentCheckpoint mergedSegmentCheckpoint = new RemoteStoreMergedSegmentCheckpoint(
             new MergedSegmentCheckpoint(
                 replicaShard.shardId(),