Clone index input in multipart upload flow

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

@@ -397,22 +397,22 @@ private void uploadBlob(
         assert ioContext != IOContext.READONCE : "Remote upload will fail with IoContext.READONCE";
         long expectedChecksum = calculateChecksumOfChecksum(from, src);
         long contentLength;
-        try (IndexInput indexInput = from.openInput(src, ioContext)) {
-            contentLength = indexInput.length();
-        }
+        IndexInput indexInput = from.openInput(src, ioContext);
+        contentLength = indexInput.length();
         boolean remoteIntegrityEnabled = false;
         if (getBlobContainer() instanceof AsyncMultiStreamBlobContainer) {
             remoteIntegrityEnabled = ((AsyncMultiStreamBlobContainer) getBlobContainer()).remoteIntegrityCheckSupported();
         }
         lowPriorityUpload = lowPriorityUpload || contentLength > ByteSizeUnit.GB.toBytes(15);
         RemoteTransferContainer.OffsetRangeInputStreamSupplier offsetRangeInputStreamSupplier;
+
         if (lowPriorityUpload) {
             offsetRangeInputStreamSupplier = (size, position) -> lowPriorityUploadRateLimiter.apply(
-                new OffsetRangeIndexInputStream(from.openInput(src, ioContext), size, position)
+                new OffsetRangeIndexInputStream(indexInput.clone(), size, position)
             );
         } else {
             offsetRangeInputStreamSupplier = (size, position) -> uploadRateLimiter.apply(
-                new OffsetRangeIndexInputStream(from.openInput(src, ioContext), size, position)
+                new OffsetRangeIndexInputStream(indexInput.clone(), size, position)
             );
         }
         RemoteTransferContainer remoteTransferContainer = new RemoteTransferContainer(
@@ -457,6 +457,14 @@ private void uploadBlob(
             }
         });
 
+        completionListener = ActionListener.runAfter(completionListener, () -> {
+            try {
+                indexInput.close();
+            } catch (IOException e) {
+                logger.warn("Error occurred while closing index input", e);
+            }
+        });
+
         WriteContext writeContext = remoteTransferContainer.createWriteContext();
         ((AsyncMultiStreamBlobContainer) blobContainer).asyncBlobUpload(writeContext, completionListener);
     }

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

@@ -599,13 +599,16 @@ public IndexInput openBlockInput(String name, long position, long length, IOCont
     public void copyFrom(Directory from, String src, IOContext context, ActionListener<Void> listener, boolean lowPriorityUpload) {
         try {
             final String remoteFileName = getNewRemoteSegmentFilename(src);
-            boolean uploaded = remoteDataDirectory.copyFrom(from, src, remoteFileName, context, () -> {
-                try {
-                    postUpload(from, src, remoteFileName, getChecksumOfLocalFile(from, src));
-                } catch (IOException e) {
-                    throw new RuntimeException("Exception in segment postUpload for file " + src, e);
-                }
-            }, listener, lowPriorityUpload);
+            boolean uploaded = false;
+            if(src.startsWith("segments_") == false) {
+                uploaded = remoteDataDirectory.copyFrom(from, src, remoteFileName, context, () -> {
+                    try {
+                        postUpload(from, src, remoteFileName, getChecksumOfLocalFile(from, src));
+                    } catch (IOException e) {
+                        throw new RuntimeException("Exception in segment postUpload for file " + src, e);
+                    }
+                }, listener, lowPriorityUpload);
+            }
             if (uploaded == false) {
                 copyFrom(from, src, src, context);
                 listener.onResponse(null);