Create abstractions to handle Remote metadata

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

@@ -17,6 +17,9 @@
 import org.apache.lucene.store.IndexInput;
 import org.apache.lucene.store.IndexOutput;
 import org.opensearch.common.UUIDs;
+import org.opensearch.index.store.metadata.RemoteSegmentMetadata;
+import org.opensearch.index.store.metadata.RemoteSegmentMetadataManager;
+import org.opensearch.index.store.metadata.SegmentMetadataParser;
 
 import java.io.IOException;
 import java.nio.file.NoSuchFileException;
@@ -75,6 +78,8 @@ public final class RemoteSegmentStoreDirectory extends FilterDirectory {
      */
     private Map<String, UploadedSegmentMetadata> segmentsUploadedToRemoteStore;
 
+    private RemoteSegmentMetadataManager remoteMetadataManager;
+
     private static final Logger logger = LogManager.getLogger(RemoteSegmentStoreDirectory.class);
 
     public RemoteSegmentStoreDirectory(RemoteDirectory remoteDataDirectory, RemoteDirectory remoteMetadataDirectory) throws IOException {
@@ -93,6 +98,7 @@ public RemoteSegmentStoreDirectory(RemoteDirectory remoteDataDirectory, RemoteDi
      */
     public void init() throws IOException {
         this.commonFilenameSuffix = UUIDs.base64UUID();
+        this.remoteMetadataManager = new RemoteSegmentMetadataManager(new SegmentMetadataParser());
         this.segmentsUploadedToRemoteStore = new ConcurrentHashMap<>(readLatestMetadataFile());
     }
 
@@ -126,10 +132,8 @@ private Map<String, UploadedSegmentMetadata> readLatestMetadataFile() throws IOE
 
     private Map<String, UploadedSegmentMetadata> readMetadataFile(String metadataFilename) throws IOException {
         try (IndexInput indexInput = remoteMetadataDirectory.openInput(metadataFilename, IOContext.DEFAULT)) {
-            Map<String, String> segmentMetadata = indexInput.readMapOfStrings();
-            return segmentMetadata.entrySet()
-                .stream()
-                .collect(Collectors.toMap(Map.Entry::getKey, entry -> UploadedSegmentMetadata.fromString(entry.getValue())));
+            RemoteSegmentMetadata metadata = this.remoteMetadataManager.readMetadata(indexInput);
+            return metadata.getMap();
         }
     }
 
@@ -138,7 +142,10 @@ private Map<String, UploadedSegmentMetadata> readMetadataFile(String metadataFil
      */
     public static class UploadedSegmentMetadata {
         // Visible for testing
+        static final int CURRENT_VERSION = 1;
+        static final String METADATA_CODEC = "segment_md";
         static final String SEPARATOR = "::";
+
         private final String originalFilename;
         private final String uploadedFilename;
         private final String checksum;
@@ -353,7 +360,7 @@ public void uploadMetadata(Collection<String> segmentFiles, Directory storeDirec
                     throw new NoSuchFileException(file);
                 }
             }
-            indexOutput.writeMapOfStrings(uploadedSegments);
+            this.remoteMetadataManager.writeMetadata(indexOutput, uploadedSegments);
             indexOutput.close();
             storeDirectory.sync(Collections.singleton(metadataFilename));
             remoteMetadataDirectory.copyFrom(storeDirectory, metadataFilename, metadataFilename, IOContext.DEFAULT);

server/src/main/java/org/opensearch/index/store/metadata/MetadataParser.java

@@ -0,0 +1,30 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.index.store.metadata;
+
+import org.apache.lucene.store.IndexInput;
+import org.apache.lucene.store.IndexOutput;
+
+import java.io.IOException;
+import java.util.Map;
+
+public interface MetadataParser<T> {
+    T readContent(IndexInput indexInput) throws IOException;
+
+    void writeContent(IndexOutput indexOutput, T content) throws IOException;
+
+    /**
+     * This method is to be removed in future and above method is supposed to be used
+     * @param indexOutput
+     * @param content
+     * @throws IOException
+     */
+    @Deprecated
+    void writeContent(IndexOutput indexOutput, Map<String, String> content) throws IOException;
+}

server/src/main/java/org/opensearch/index/store/metadata/RemoteSegmentMetadata.java

@@ -0,0 +1,47 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.index.store.metadata;
+
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import org.opensearch.index.store.RemoteSegmentStoreDirectory;
+
+public class RemoteSegmentMetadata {
+    private final Map<String, RemoteSegmentStoreDirectory.UploadedSegmentMetadata> metadata;
+
+    public RemoteSegmentMetadata(Map<String, RemoteSegmentStoreDirectory.UploadedSegmentMetadata> metadata) {
+        this.metadata = metadata;
+    }
+
+    public Map<String, RemoteSegmentStoreDirectory.UploadedSegmentMetadata> getMetadata() {
+        return this.metadata;
+    }
+
+    public static RemoteSegmentMetadata fromMapOfStrings(Map<String, String> segmentMetadata) {
+        return new RemoteSegmentMetadata(
+            segmentMetadata.entrySet()
+                .stream()
+                .collect(
+                    Collectors.toMap(
+                        Map.Entry::getKey,
+                        entry -> RemoteSegmentStoreDirectory.UploadedSegmentMetadata.fromString(entry.getValue())
+                    )
+                )
+        );
+    }
+
+    /**
+     * Ideally we shouldn't need expose internal data structures. all operations should be added into this class
+     * @return
+     */
+    public Map<String, RemoteSegmentStoreDirectory.UploadedSegmentMetadata> getMap() {
+        return this.metadata;
+    }
+}

server/src/main/java/org/opensearch/index/store/metadata/RemoteSegmentMetadataManager.java

@@ -0,0 +1,66 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.index.store.metadata;
+
+import java.io.IOException;
+import java.util.Map;
+
+import org.apache.lucene.codecs.CodecUtil;
+import org.apache.lucene.store.BufferedChecksumIndexInput;
+import org.apache.lucene.store.ChecksumIndexInput;
+import org.apache.lucene.store.IndexInput;
+import org.apache.lucene.store.IndexOutput;
+
+public class RemoteSegmentMetadataManager {
+    static final int CURRENT_VERSION = 1;
+    static final String METADATA_CODEC = "segment_md";
+
+    // replace with a parser factory to support multiple versions of metadata in future if needed
+    private final MetadataParser<RemoteSegmentMetadata> parser;
+
+    public RemoteSegmentMetadataManager(MetadataParser<RemoteSegmentMetadata> parser) {
+        this.parser = parser;
+    }
+
+    public RemoteSegmentMetadata readMetadata(IndexInput indexInput) throws IOException {
+        ChecksumIndexInput checksumIndexInput = new BufferedChecksumIndexInput(indexInput);
+        checkHeader(checksumIndexInput);
+        RemoteSegmentMetadata metadata = this.parser.readContent(checksumIndexInput);
+        checkFooter(checksumIndexInput);
+        return metadata;
+    }
+
+    /**
+     * this method should only accept RemoteSegmentMetadata and not Map<String, String>
+     * @param indexOutput
+     * @param metadata
+     * @throws IOException
+     */
+    public void writeMetadata(IndexOutput indexOutput, Map<String, String> metadata) throws IOException {
+        this.writeHeader(indexOutput);
+        this.parser.writeContent(indexOutput, metadata);
+        this.writeFooter(indexOutput);
+    }
+
+    private int checkHeader(IndexInput indexInput) throws IOException {
+        return CodecUtil.checkHeader(indexInput, METADATA_CODEC, CURRENT_VERSION, CURRENT_VERSION);
+    }
+
+    private void checkFooter(ChecksumIndexInput indexInput) throws IOException {
+        CodecUtil.checkFooter(indexInput);
+    }
+
+    private void writeHeader(IndexOutput indexOutput) throws IOException {
+        CodecUtil.writeHeader(indexOutput, METADATA_CODEC, CURRENT_VERSION);
+    }
+
+    private void writeFooter(IndexOutput indexOutput) throws IOException {
+        CodecUtil.writeFooter(indexOutput);
+    }
+}

server/src/main/java/org/opensearch/index/store/metadata/SegmentMetadataParser.java

@@ -0,0 +1,30 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.index.store.metadata;
+
+import java.io.IOException;
+import java.util.Map;
+
+import org.apache.lucene.store.IndexInput;
+import org.apache.lucene.store.IndexOutput;
+
+public class SegmentMetadataParser implements MetadataParser<RemoteSegmentMetadata> {
+    @Override
+    public RemoteSegmentMetadata readContent(IndexInput indexInput) throws IOException {
+        return RemoteSegmentMetadata.fromMapOfStrings(indexInput.readMapOfStrings());
+    }
+
+    @Override
+    public void writeContent(IndexOutput indexOutput, RemoteSegmentMetadata content) throws IOException {}
+
+    @Override
+    public void writeContent(IndexOutput indexOutput, Map<String, String> content) throws IOException {
+        indexOutput.writeMapOfStrings(content);
+    }
+}