Implement range reads in HDFS repository

CHANGELOG.md

@@ -89,6 +89,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - [BWC and API enforcement] Define the initial set of annotations, their meaning and relations between them ([#9223](https://github.com/opensearch-project/OpenSearch/pull/9223))
 - [Segment Replication] Support realtime reads for GET requests ([#9212](https://github.com/opensearch-project/OpenSearch/pull/9212))
 - [Feature] Expose term frequency in Painless script score context ([#9081](https://github.com/opensearch-project/OpenSearch/pull/9081))
+- Add support for reading partial files to HDFS repository ([#9513](https://github.com/opensearch-project/OpenSearch/issues/9513))
 
 ### Dependencies
 - Bump `org.apache.logging.log4j:log4j-core` from 2.17.1 to 2.20.0 ([#8307](https://github.com/opensearch-project/OpenSearch/pull/8307))

plugins/repository-hdfs/src/main/java/org/opensearch/repositories/hdfs/HdfsBlobContainer.java

@@ -32,6 +32,7 @@
 package org.opensearch.repositories.hdfs;
 
 import org.apache.hadoop.fs.CreateFlag;
+import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileContext;
 import org.apache.hadoop.fs.FileStatus;
@@ -46,6 +47,7 @@
 import org.opensearch.common.blobstore.fs.FsBlobContainer;
 import org.opensearch.common.blobstore.support.AbstractBlobContainer;
 import org.opensearch.common.blobstore.support.PlainBlobMetadata;
+import org.opensearch.common.io.Streams;
 import org.opensearch.repositories.hdfs.HdfsBlobStore.Operation;
 
 import java.io.FileNotFoundException;
@@ -125,8 +127,23 @@
     }
 
     @Override
-    public InputStream readBlob(String blobName, long position, long length) {
-        throw new UnsupportedOperationException();
+    public InputStream readBlob(String blobName, long position, long length) throws IOException {
+        return store.execute(fileContext -> {
+            final FSDataInputStream stream;
+            try {
+                stream = fileContext.open(new Path(path, blobName), bufferSize);
+            } catch (FileNotFoundException fnfe) {
+                throw new NoSuchFileException("[" + blobName + "] blob not found");
+            }
+            // Seek to the desired start position, closing the stream if any error occurs
+            try {
+                stream.seek(position);
+            } catch (Exception e) {
+                stream.close();
+                throw e;
+            }
+            return Streams.limitStream(new HDFSPrivilegedInputSteam(stream, securityContext), length);
+        });
     }
 
     @Override

plugins/repository-hdfs/src/test/java/org/opensearch/repositories/hdfs/HdfsBlobStoreRepositoryTests.java

@@ -66,8 +66,4 @@ protected Settings repositorySettings() {
     protected Collection<Class<? extends Plugin>> nodePlugins() {
         return Collections.singletonList(HdfsPlugin.class);
     }
-
-    @AwaitsFix(bugUrl = "https://github.com/opensearch-project/OpenSearch/issues/9513")
-    @Override
-    public void testReadRange() {}
 }