Rename new constructs and minor refactoring around version specific reading contract

Signed-off-by: Varun Bansal <bansvaru@amazon.com>

Author: linuxpi

server/src/main/java/org/opensearch/common/io/StreamReadWriteHandler.java

@@ -0,0 +1,33 @@
+/*
+ * 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.common.io;
+
+import org.apache.lucene.store.IndexInput;
+import org.apache.lucene.store.IndexOutput;
+
+import java.io.IOException;
+
+/**
+ * Interface for reading/writing content streams to/from - {@link T}
+ * @param <T> The type of content to be read/written to stream
+ */
+public interface StreamReadWriteHandler<T> {
+    /**
+     * Implements logic to read content from file input stream {@code indexInput} and parse into {@link T}
+     * @param indexInput file input stream
+     * @return content parsed to {@link T}
+     */
+    T readContent(IndexInput indexInput) throws IOException;
+
+    /**
+     * Implements logic to write content from {@code content} to file output stream {@code indexOutput}
+     * @param indexOutput file input stream
+     */
+    void writeContent(IndexOutput indexOutput, T content) throws IOException;
+}

server/src/main/java/org/opensearch/common/io/VersionedCodecStreamWrapper.java

@@ -0,0 +1,108 @@
+/*
+ * 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.common.io;
+
+import java.io.IOException;
+
+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;
+
+/**
+ * Manages versioning and checksum for a stream of content.
+ * @param <T> Type of content to be read/written
+ */
+public class VersionedCodecStreamWrapper<T> {
+    // This can be updated to hold a parserFactory and get relevant parsers based on the stream versions
+    private final StreamReadWriteHandler<T> parser;
+    private final int currentVersion;
+    private final String codec;
+
+    /**
+     * @param parser parser to read/write stream from T
+     * @param currentVersion latest supported version of the stream
+     * @param codec: stream codec
+     */
+    public VersionedCodecStreamWrapper(StreamReadWriteHandler<T> parser, int currentVersion, String codec) {
+        this.parser = parser;
+        this.currentVersion = currentVersion;
+        this.codec = codec;
+    }
+
+    /**
+     * Reads stream content from {@code indexInput} and parses the read content to {@link T}.
+     * Before reading actual content, verifies the header with relevant codec and version.
+     * After reading the actual content, verifies the checksum as well
+     * @param indexInput file input stream
+     * @return stream content parsed into {@link T}
+     */
+    public T readStream(IndexInput indexInput) throws IOException {
+        ChecksumIndexInput checksumIndexInput = new BufferedChecksumIndexInput(indexInput);
+        int readStreamVersion = checkHeader(checksumIndexInput);
+        T content = getParserForVersion(readStreamVersion).readContent(checksumIndexInput);
+        checkFooter(checksumIndexInput);
+        return content;
+    }
+
+    /**
+     * Writes to file output stream {@code indexOutput}
+     * @param indexOutput file output stream which will store stream content
+     * @param content stream content.
+     */
+    public void writeStream(IndexOutput indexOutput, T content) throws IOException {
+        this.writeHeader(indexOutput);
+        getParserForVersion(this.currentVersion).writeContent(indexOutput, content);
+        this.writeFooter(indexOutput);
+    }
+
+    /**
+     * Reads header from file input stream containing {@code this.codec} and {@code this.currentVersion}.
+     * @param indexInput file input stream
+     * @return header version found in the input stream
+     */
+    private int checkHeader(IndexInput indexInput) throws IOException {
+        return CodecUtil.checkHeader(indexInput, this.codec, this.currentVersion, this.currentVersion);
+    }
+
+    /**
+     * Reads footer from file input stream containing checksum.
+     * The {@link IndexInput#getFilePointer()} should be at the footer start position.
+     * @param indexInput file input stream
+     */
+    private void checkFooter(ChecksumIndexInput indexInput) throws IOException {
+        CodecUtil.checkFooter(indexInput);
+    }
+
+    /**
+     * Writes header with {@code this.codec} and {@code this.currentVersion} to the file output stream
+     * @param indexOutput file output stream
+     */
+    private void writeHeader(IndexOutput indexOutput) throws IOException {
+        CodecUtil.writeHeader(indexOutput, this.codec, this.currentVersion);
+    }
+
+    /**
+     * Writes footer with checksum of contents of file output stream
+     * @param indexOutput file output stream
+     */
+    private void writeFooter(IndexOutput indexOutput) throws IOException {
+        CodecUtil.writeFooter(indexOutput);
+    }
+
+    /**
+     * Returns relevant parser for the version
+     * @param version stream content version
+     */
+    private StreamReadWriteHandler<T> getParserForVersion(int version) {
+        // TODO implement parser factory and pick relevant parser based on version
+        return this.parser;
+    }
+}

server/src/main/java/org/opensearch/common/metadata/MetadataContentParser.java

@@ -18,7 +18,8 @@
 import org.apache.lucene.store.IndexOutput;
 import org.opensearch.common.UUIDs;
 import org.opensearch.index.store.remote.metadata.RemoteSegmentMetadata;
-import org.opensearch.common.metadata.MetadataParser;
+import org.opensearch.common.io.VersionedCodecStreamWrapper;
+import org.opensearch.index.store.remote.metadata.RemoteSegmentMetadataHandler;
 
 import java.io.IOException;
 import java.nio.file.NoSuchFileException;
@@ -77,19 +78,18 @@ public final class RemoteSegmentStoreDirectory extends FilterDirectory {
      */
     private Map<String, UploadedSegmentMetadata> segmentsUploadedToRemoteStore;
 
-    private MetadataParser<RemoteSegmentMetadata> remoteMetadataParser;
+    private static final VersionedCodecStreamWrapper<RemoteSegmentMetadata> metadataStreamWrapper = new VersionedCodecStreamWrapper<>(
+        new RemoteSegmentMetadataHandler(),
+        RemoteSegmentMetadata.CURRENT_VERSION,
+        RemoteSegmentMetadata.METADATA_CODEC
+    );
 
     private static final Logger logger = LogManager.getLogger(RemoteSegmentStoreDirectory.class);
 
-    public RemoteSegmentStoreDirectory(
-        RemoteDirectory remoteDataDirectory,
-        RemoteDirectory remoteMetadataDirectory,
-        MetadataParser<RemoteSegmentMetadata> metadataParser
-    ) throws IOException {
+    public RemoteSegmentStoreDirectory(RemoteDirectory remoteDataDirectory, RemoteDirectory remoteMetadataDirectory) throws IOException {
         super(remoteDataDirectory);
         this.remoteDataDirectory = remoteDataDirectory;
         this.remoteMetadataDirectory = remoteMetadataDirectory;
-        this.remoteMetadataParser = metadataParser;
         init();
     }
 
@@ -135,7 +135,7 @@ private Map<String, UploadedSegmentMetadata> readLatestMetadataFile() throws IOE
 
     private Map<String, UploadedSegmentMetadata> readMetadataFile(String metadataFilename) throws IOException {
         try (IndexInput indexInput = remoteMetadataDirectory.openInput(metadataFilename, IOContext.DEFAULT)) {
-            RemoteSegmentMetadata metadata = this.remoteMetadataParser.readMetadata(indexInput);
+            RemoteSegmentMetadata metadata = metadataStreamWrapper.readStream(indexInput);
             return metadata.getMetadata();
         }
     }
@@ -361,7 +361,7 @@ public void uploadMetadata(Collection<String> segmentFiles, Directory storeDirec
                     throw new NoSuchFileException(file);
                 }
             }
-            this.remoteMetadataParser.writeMetadata(indexOutput, RemoteSegmentMetadata.fromMapOfStrings(uploadedSegments));
+            metadataStreamWrapper.writeStream(indexOutput, RemoteSegmentMetadata.fromMapOfStrings(uploadedSegments));
             indexOutput.close();
             storeDirectory.sync(Collections.singleton(metadataFilename));
             remoteMetadataDirectory.copyFrom(storeDirectory, metadataFilename, metadataFilename, IOContext.DEFAULT);

server/src/main/java/org/opensearch/common/metadata/MetadataParser.java

@@ -13,7 +13,6 @@
 import org.opensearch.common.blobstore.BlobPath;
 import org.opensearch.index.IndexSettings;
 import org.opensearch.index.shard.ShardPath;
-import org.opensearch.index.store.remote.metadata.RemoteSegmentMetadata;
 import org.opensearch.plugins.IndexStorePlugin;
 import org.opensearch.repositories.RepositoriesService;
 import org.opensearch.repositories.Repository;
@@ -48,7 +47,7 @@ public Directory newDirectory(String repositoryName, IndexSettings indexSettings
             RemoteDirectory dataDirectory = createRemoteDirectory(repository, commonBlobPath, "data");
             RemoteDirectory metadataDirectory = createRemoteDirectory(repository, commonBlobPath, "metadata");
 
-            return new RemoteSegmentStoreDirectory(dataDirectory, metadataDirectory, RemoteSegmentMetadata.createMetadataParser());
+            return new RemoteSegmentStoreDirectory(dataDirectory, metadataDirectory);
         } catch (RepositoryMissingException e) {
             throw new IllegalArgumentException("Repository should be created before creating index with remote_store enabled setting", e);
         }

server/src/main/java/org/opensearch/common/metadata/package-info.java

@@ -11,7 +11,7 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 
-import org.opensearch.common.metadata.MetadataParser;
+import org.opensearch.common.io.VersionedCodecStreamWrapper;
 import org.opensearch.index.store.RemoteSegmentStoreDirectory;
 
 /**
@@ -71,10 +71,10 @@ public static RemoteSegmentMetadata fromMapOfStrings(Map<String, String> segment
     }
 
     /**
-     * Constructs {@link MetadataParser} for {@link RemoteSegmentMetadata}
-     * @return {@link MetadataParser}
+     * Constructs {@link VersionedCodecStreamWrapper} for {@link RemoteSegmentMetadata}
+     * @return {@link VersionedCodecStreamWrapper}
      */
-    public static MetadataParser<RemoteSegmentMetadata> createMetadataParser() {
-        return new MetadataParser<>(new RemoteSegmentMetadataContentParser(), CURRENT_VERSION, METADATA_CODEC);
+    public static VersionedCodecStreamWrapper<RemoteSegmentMetadata> createMetadataHandler() {
+        return new VersionedCodecStreamWrapper<>(new RemoteSegmentMetadataHandler(), CURRENT_VERSION, METADATA_CODEC);
     }
 }

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

@@ -12,12 +12,12 @@
 
 import org.apache.lucene.store.IndexInput;
 import org.apache.lucene.store.IndexOutput;
-import org.opensearch.common.metadata.MetadataContentParser;
+import org.opensearch.common.io.StreamReadWriteHandler;
 
 /**
  * Parser for {@link RemoteSegmentMetadata}
  */
-public class RemoteSegmentMetadataContentParser implements MetadataContentParser<RemoteSegmentMetadata> {
+public class RemoteSegmentMetadataHandler implements StreamReadWriteHandler<RemoteSegmentMetadata> {
     /**
      * Reads metadata content from metadata file input stream and parsed into {@link RemoteSegmentMetadata}
      * @param indexInput metadata file input stream with {@link IndexInput#getFilePointer()} pointing to metadata content