Multi-part parallel download of large segment files from remote store

[Bukhtawar]

Description

Adds multi-part parallel download of large segment files from the remote store, for segment-replication replicas and remote-store recoveries.

Today. RemoteStoreFileDownloader already parallelises across files (bounded by indices.recovery.max_concurrent_remote_store_streams), but each file is one sequential stream via Directory.copyFrom. A large segment file (.fdt, .doc, .dvd, .cfs, ...) is therefore the long pole of every transfer: once the small files are done, the remaining streams sit idle while one thread drains one object.

History. A multi-stream-within-a-file download existed once and was removed in #10519 because it wrote parts positionally into a temp file, bypassing the Lucene Directory pipeline. That broke recovery stats, Lucene checksum verification and (client-side) encryption integration. With server-side encryption now the norm, the encryption concern no longer applies, and this PR is designed so the other two never arise.

Why the #10519 problems do not come back. Both broke because the parts were written positionally via FileChannel, so nothing hooked on Directory#copyFrom / IndexOutput ever saw the bytes. Here RemoteStoreFileDownloader still calls destination.copyFrom(source, file, file, ctx) unchanged; only what source.openInput(file) returns differs. Consequently:

• Recovery stats — ReplicationStatsDirectoryWrapper (counts readBytes on the source input to feed the progress tracker) and Store.StoreDirectory#copyFrom (the DirectoryFileTransferTracker behind remote_store.download node stats) wrap the same call and see every byte, incrementally, including the failure path.
• Checksum verification — the bytes are presented in order and written through a normal IndexOutput, so Store#createVerifyingOutput / Store#verify (the MultiFileWriter pattern) work with no re-read; a byte flipped inside a prefetched part is caught by the running CRC, not just by the footer. The default remote-store download path does not verify at write time today and this PR keeps that posture unchanged — it makes write-time verification possible, not mandatory.

Both properties are asserted end-to-end, each with a negative control, in RemoteStoreMultiPartDownloadPipelineTests.

Approach — parallel fetch, ordered write.

• ParallelPartInputStream is a sequential InputStream over a blob. The part currently being read is streamed directly (no buffering); the following parts are prefetched on the remote_recovery pool into heap buffers via BlobContainer#readBlob(name, position, length). Plain byte-range reads work on every blob store and return plaintext under SSE — no dependence on how the object was multipart-uploaded.
• The stream is wrapped in RemoteIndexInput, so the unchanged Directory#copyFrom path consumes it. ReplicationStatsDirectoryWrapper, Store.StoreDirectory transfer stats, local directory-level encryption and IndexOutput-side checksum verification all keep working because the write side is still a plain sequential IndexOutput.
• Deadlock-free by construction. The reader never waits on work that has not started: a part whose prefetch task is still queued is stolen and read by the reader, and the queued task becomes a no-op. With no permits the stream degrades to sequential range-by-range reads. It is therefore safe to drive from the same remote_recovery pool that runs the prefetches.
• ParallelDownloadPermits is a node-wide, dynamically resizable budget of in-flight prefetched parts shared by all downloads on the node. Heap held by parallel downloads is bounded by max_concurrent_parts × part_size (128 MB on a 16-vCPU node with defaults).
• RemoteStoreFileDownloader wraps a RemoteSegmentStoreDirectory source in a FilterDirectory whose openInput returns the multi-part stream for files larger than one part. DataFormatAwareRemoteDirectory routes the range reads to the container of the blob's data format.

New dynamic node settings

Setting	Default	Notes
indices.recovery.remote_store.parallel_download.part_size	16mb	range 1mb..1gb; files ≤ one part keep the single-stream path
indices.recovery.remote_store.parallel_download.max_concurrent_parts	max(1, allocated_processors / 2)	node-wide prefetch budget; 0 disables the feature

Tests

• ParallelPartInputStreamTests — byte-exact ordering across parts, single-byte reads, permit accounting under a gated executor, reader stealing queued prefetches (and stolen tasks being no-ops), RejectedExecutionException fallback, prefetch-failure propagation with permit release, short-range-read detection, close() returning outstanding permits, rate-limiter wrapping every part.
• ParallelDownloadPermitsTests — acquire/release, zero budget, grow, shrink-takes-effect-as-released, negative rejected.
• RemoteStoreFileDownloaderTests — end-to-end over a real RemoteSegmentStoreDirectory + FS blob container: multi-part files land byte-identical with valid Lucene footers (CodecUtil.checksumEntireFile), exactly one range request per part, small files still use the single-stream path, all permits returned; and the max_concurrent_parts = 0 disabled path.
• RecoverySettingsDynamicUpdateTests — defaults, dynamic updates, bounds, and that the permit budget is resized in place (it is shared with in-flight downloads).
• RemoteStoreMultiPartDownloadPipelineTests — end-to-end over a real RemoteSegmentStoreDirectory with a multi-part file and a single-stream file: (a) ReplicationStatsDirectoryWrapper progress equals every file's length and is reported incrementally, and Store's DirectoryFileTransferTracker records started/succeeded bytes — plus a negative control where an injected range-read failure lands in failed_bytes; (b) Store#createVerifyingOutput + Store#verify pass on the multi-part stream with exactly one range request per part and no re-read — plus a negative control where one byte flipped inside a prefetched part (footer intact) is rejected with CorruptIndexException.

:server:precommit passes.

Benchmark (EC2, S3 remote store)

Two 2-node clusters (r6g.2xlarge, 8 vCPU, ~2.5 Gbps baseline network; gp3 @ 1000 MB/s), 1 primary + 1 replica, segrep + remote store on S3 (us-east-1), same binary (this PR's head) on both. The only difference between arms is the dynamic setting indices.recovery.remote_store.parallel_download.max_concurrent_parts — 0 is exactly today's code path. max_concurrent_remote_store_streams=4, part_size=16mb, rate limiters disabled. Replication lag sampled from _cat/segment_replication every 5 s (phase 1) / 2 s (phase 2). Zero replication failures in any run.

Phase 1 — refresh-dominated (geonames, 11.4 M docs, default compound ratio, refresh_interval=30s)

arm	rounds	round p50	round p90	largest round	effective
parts=0	10	3.38 s	4.18 s	995 MB in 4.56 s	218 MB/s
parts=0 (repeat)	9	3.61 s	4.20 s	1,512 MB	—
parts=4	9	3.64 s	4.95 s	1,644 MB in 7.81 s	211 MB/s
parts=8	10	3.06 s	4.83 s	1,324 MB in 5.52 s	240 MB/s

Neutral. Large merged segments were non-compound (~10 files each), so the existing cross-file fan-out already saturated the instance NIC; splitting files further cannot add throughput to a full pipe, and it did not regress anything.

Phase 2 — merge-heavy, index.compound_format: true (2 × geonames = 22.8 M docs into one shard, then _forcemerge?max_num_segments=1 → a single 5.3 GB .cfs)

arm	rounds	round p50	round p90	5.3 GB .cfs round	effective	replica download rate while busy
parts=0	19	4.65 s	14.7 s	58.6 s	91 MB/s	95 MB/s
parts=4 (default on 8 vCPU)	20	5.08 s	9.7 s	39.1 s	136 MB/s	167 MB/s
parts=8	21	3.78 s	7.0 s	28.5 s	186 MB/s	255 MB/s

When a replication round is dominated by one large file — a compound-format merged segment here — the baseline is bound to a single S3 connection (~90 MB/s). Multi-part download raises it to 136 MB/s at 4 parts and 186 MB/s at 8, i.e. ~2× faster catch-up on the 5.3 GB segment and the p90 round time halved (14.7 s → 7.0 s). The intermediate 1.1–1.5 GB .cfs merges show the same pattern (parts=0: 13.5 / 14.7 / 14.8 s; parts=4: 9.7 / 10.0 s; parts=8: 7.0 / 8.9 s). Replica-side remote_store.download stats agree: ~21 GB downloaded per run in 222 s of download time at parts=0 vs 81 s at parts=8. At parts=8 the replica was at ~255 MB/s, close to the r6g.2xlarge network allowance, so the gain is NIC-bounded on this instance size.

Caveats: arms ran sequentially on one load generator (opensearch-benchmark allows one run per host); one run per arm in phase 2; round time includes local commit and reader reopen, not only the download. Heap on the replica was not sampled — the budget bound is max_concurrent_parts × part_size (128 MB at parts=8).

Summary: neutral when the replica's network is already saturated by cross-file parallelism; up to ~2× faster on rounds dominated by a single large file (compound-format segments, large .fdt/.dvd-heavy merges), with no failures or regressions observed.

Related Issues

no linked issue: follow-up to the design discussion in #10519, which removed the previous multi-stream download; no tracking issue exists yet.

Check List

• Functionality includes testing.
• API changes companion pull request created, if applicable.
• Public documentation issue/PR created, if applicable. (Two new node settings — documentation PR to follow once out of draft.)

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit 89bc4fc)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 No multiple PR themes

⚡ Recommended focus areas for review Possible Issue In close(), prefetches whose task has already started (owner != NONE) have their permit released via future.whenComplete. However, if the Prefetch.run() method sets the owner to PREFETCHER and then completes exceptionally, releasePermit() is only called by the whenComplete callback registered in close() — but the normal completion path in run() never releases the permit itself. This is by design (the reader releases when it consumes), but if the stream is closed after a running prefetch has completed successfully but before the reader consumed it, the buffer is discarded and only the whenComplete callback releases the permit. This works, but note that Prefetch.run() does not release the permit on the success path — only the reader (via finishCurrentPart) or close() does. If the reader errors out mid-part (e.g., the EOFException path in read), finishCurrentPart is not called for the current part before the exception propagates, potentially leaving currentPrefetch's permit unreleased until close() is invoked. Verify callers always close the stream on error. public void close() throws IOException { if (closed.compareAndSet(false, true) == false) { return; } IOException failure = null; try { finishCurrentPart(); } catch (IOException e) { failure = e; } // Return every permit this stream holds. A part that has not started is cancelled and released now. A part // whose task is already running still owns a buffer of up to partSize bytes until it completes, so its permit // is released only when the future completes (immediately if it already has) -- that keeps the node-wide // permits * partSize bound exact instead of letting the buffer briefly outlive its permit. for (Prefetch prefetch : prefetches.values()) { if (prefetch.owner.compareAndSet(Owner.NONE, Owner.CANCELLED)) { prefetch.releasePermit(); } else { prefetch.future.whenComplete((bytes, t) -> prefetch.releasePermit()); } } prefetches.clear(); if (failure != null) { throw failure; } } Possible Issue schedulePrefetches uses permits.getMaxPermits() to bound lookahead, but maxPermits may be reduced dynamically at runtime. If maxPermits is later shrunk below the number of already-scheduled prefetches (tracked by highestScheduledPart), the loop still works, but if it is grown while a stream is active, highestScheduledPart prevents re-considering earlier parts. This is acceptable, but the more subtle concern: nextPart + permits.getMaxPermits() can overflow if maxPermits is set to Integer.MAX_VALUE. Consider using saturating arithmetic or bounding maxPermits. private void schedulePrefetches() { if (permits == null) { return; } // Do not read further ahead than the budget would ever allow. final int lookaheadLimit = Math.min(numParts, nextPart + permits.getMaxPermits()); for (int partIndex = Math.max(nextPart, highestScheduledPart + 1); partIndex < lookaheadLimit; partIndex++) { if (permits.tryAcquire() == false) { return; } final Prefetch prefetch = new Prefetch(partIndex); prefetches.put(partIndex, prefetch); highestScheduledPart = partIndex; try { executor.execute(prefetch); } catch (RejectedExecutionException e) { prefetches.remove(partIndex); prefetch.owner.set(Owner.CANCELLED); prefetch.releasePermit(); final int rejectedPart = partIndex; logger.debug(() -> new ParameterizedMessage("Prefetch of part {} for blob [{}] rejected", rejectedPart, blobName), e); return; } } } Potential Behavior Change wrapForParallelPartDownload wraps the source in a FilterDirectory that overrides openInput, but copyOneFile may call other Directory methods on the source (e.g., fileLength). The wrapper delegates to in, which is fine, but any callers casting the source to RemoteSegmentStoreDirectory after this wrapping would fail. Verify that copyOneFile and downstream code only use the Directory interface on the wrapped source. private Directory wrapForParallelPartDownload(Directory source) { if (source instanceof RemoteSegmentStoreDirectory == false) { return source; } final ParallelDownloadPermits permits = recoverySettings.getRemoteStoreParallelDownloadPermits(); if (permits.getMaxPermits() <= 0) { return source; } final RemoteSegmentStoreDirectory remoteSource = (RemoteSegmentStoreDirectory) source; final long partSize = recoverySettings.getRemoteStoreParallelDownloadPartSize().getBytes(); final Executor executor = threadPool.executor(ThreadPool.Names.REMOTE_RECOVERY); return new FilterDirectory(source) { @Override public IndexInput openInput(String name, IOContext context) throws IOException { final long fileLength = in.fileLength(name); if (fileLength <= partSize) { return in.openInput(name, context); } logger.trace("Downloading file {} of size {} in parts of {} bytes", name, fileLength, partSize); return remoteSource.openParallelInput(name, fileLength, context, partSize, executor, permits); } }; }

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 89bc4fc

Explore these optional code suggestions:

Category	Suggestion	Impact
General	Remove misleading infinite loop wrapper The while (true) loop unconditionally returns after one iteration, so the loop construct is misleading and dead. More importantly, if openNextPart() opens a zero-length part (which cannot happen given partLength guarantees, but defensively) or if the caller passes len == 0, behavior would be inconsistent. Consider removing the loop wrapper since control flow always exits on the first pass, making the code clearer. server/src/main/java/org/opensearch/index/store/ParallelPartInputStream.java [129-159] -while (true) { - if (current == null) { - if (nextPart >= numParts) { - return -1; - } - openNextPart(); +if (current == null) { + if (nextPart >= numParts) { + return -1; } - final int toRead = (int) Math.min(len, currentRemaining); - final int n = current.read(b, off, toRead); + openNextPart(); +} +final int toRead = (int) Math.min(len, currentRemaining); +final int n = current.read(b, off, toRead); Suggestion importance[1-10]: 3 __ Why: The while (true) loop is indeed unreachable beyond one iteration and could be simplified for readability, but this is a minor style issue with no functional impact.	Low
	Release prefetch buffers promptly on close When closing while a prefetch is running and has already completed successfully, the buffer produced by future.get() is never consumed and remains reachable via the completed future until GC. Additionally, if whenComplete runs synchronously on the executor and the task already failed, this is fine — but if the task is still running and produces a large buffer, the buffer will be held until the task completes. Consider explicitly discarding/nulling the future result on close to release the buffer sooner. server/src/main/java/org/opensearch/index/store/ParallelPartInputStream.java [177-183] for (Prefetch prefetch : prefetches.values()) { if (prefetch.owner.compareAndSet(Owner.NONE, Owner.CANCELLED)) { prefetch.releasePermit(); } else { - prefetch.future.whenComplete((bytes, t) -> prefetch.releasePermit()); + prefetch.future.whenComplete((bytes, t) -> { + prefetch.releasePermit(); + // buffer, if any, becomes unreachable when the future is GCed + }); } } Suggestion importance[1-10]: 2 __ Why: The suggestion's improved_code is essentially identical to the existing_code (only adds a comment), and does not actually address the concern raised about releasing buffers sooner.	Low
	Avoid overflow in part count computation When fileLength == 0, numParts becomes 0 and read() correctly returns -1 on first call. However, fileLength + partSize - 1 could overflow to a negative value if fileLength is near Long.MAX_VALUE. While unrealistic for segment files, consider computing this more safely, e.g. fileLength == 0 ? 0 : (fileLength - 1) / partSize + 1, to avoid the overflow edge case. server/src/main/java/org/opensearch/index/store/ParallelPartInputStream.java [106] -this.numParts = Math.toIntExact((fileLength + partSize - 1) / partSize); +this.numParts = fileLength == 0 ? 0 : Math.toIntExact((fileLength - 1) / partSize + 1); Suggestion importance[1-10]: 2 __ Why: The overflow concern is theoretically valid but unrealistic for segment file sizes; the suggestion itself acknowledges this. Impact is very low.	Low

---

Previous suggestions

Suggestions up to commit 00bcf7f

Category	Suggestion	Impact
General	Avoid thread pool saturation between prefetch and copy The FilterDirectory overrides openInput but keeps in as the original source. If remoteSource is unwrapped from a wrapper, calling in.fileLength(name) and in.openInput(name, context) may bypass the wrapper's behavior. Since remoteSource is the same as source (only cast), this is fine here — however, calling openParallelInput from within openInput occurs on the thread executing copyOneFile, which is the REMOTE_RECOVERY pool: prefetch tasks are submitted to the same pool, risking pool saturation/deadlock if the pool is small. Consider using a separate executor or documenting/enforcing that the pool size exceeds max_concurrent_remote_store_streams + max_concurrent_parts. server/src/main/java/org/opensearch/index/store/RemoteStoreFileDownloader.java [155-165] return new FilterDirectory(source) { @Override public IndexInput openInput(String name, IOContext context) throws IOException { final long fileLength = in.fileLength(name); if (fileLength <= partSize) { return in.openInput(name, context); } logger.trace("Downloading file {} of size {} in parts of {} bytes", name, fileLength, partSize); + // NOTE: prefetches run on the same REMOTE_RECOVERY pool as the caller; ensure pool sizing accommodates both. return remoteSource.openParallelInput(name, fileLength, context, partSize, executor, permits); } }; Suggestion importance[1-10]: 6 __ Why: Highlights a legitimate concern about running prefetch tasks and copy tasks on the same REMOTE_RECOVERY pool, which could risk saturation. However, the stream is designed to fall back gracefully when tasks can't run, mitigating deadlock risk. The suggested change is only a comment.	Low
	Avoid integer overflow in lookahead computation nextPart + permits.getMaxPermits() can overflow to a negative value when nextPart is large (numParts can be close to Integer.MAX_VALUE), causing Math.min to return that negative value and effectively disable prefetching. Compute the lookahead using long arithmetic or clamp before summing. server/src/main/java/org/opensearch/index/store/ParallelPartInputStream.java [252-257] private void schedulePrefetches() { if (permits == null) { return; } // Do not read further ahead than the budget would ever allow. - final int lookaheadLimit = Math.min(numParts, nextPart + permits.getMaxPermits()); + final long lookaheadLong = (long) nextPart + permits.getMaxPermits(); + final int lookaheadLimit = (int) Math.min(numParts, lookaheadLong); Suggestion importance[1-10]: 5 __ Why: Overflow in nextPart + permits.getMaxPermits() is theoretically possible with extremely large numParts, though unlikely in practice. Using long arithmetic is a reasonable defensive fix.	Low
	Validate read parameters per InputStream contract The read(byte[], int, int) method does not validate the b, off, len parameters as required by the InputStream contract. Add explicit null and bounds checks (or delegate to Objects.checkFromIndexSize) to throw NullPointerException/IndexOutOfBoundsException early, matching standard InputStream behavior and preventing subtle downstream corruption. server/src/main/java/org/opensearch/index/store/ParallelPartInputStream.java [124-135] @Override public int read(byte[] b, int off, int len) throws IOException { + java.util.Objects.checkFromIndexSize(off, len, b.length); ensureOpen(); if (len == 0) { return 0; } while (true) { if (current == null) { if (nextPart >= numParts) { return -1; } openNextPart(); } Suggestion importance[1-10]: 3 __ Why: Adding bounds checks aligns with the InputStream contract but is a minor robustness improvement; underlying current.read already performs some validation. Low impact.	Low
Possible issue	Enforce non-null constructor arguments consistently schedulePrefetches unconditionally dereferences executor, and Prefetch.releasePermit() calls permits.release() even though permits may be null. The class documents that a null permits degrades to sequential reads but does not enforce non-null executor, and permits==null will NPE inside close/prefetch flows. Either forbid null (validate in the constructor) or handle both null cases consistently. server/src/main/java/org/opensearch/index/store/ParallelPartInputStream.java [87-110] public ParallelPartInputStream( BlobContainer blobContainer, String blobName, long fileLength, long partSize, Executor executor, ParallelDownloadPermits permits, UnaryOperator<InputStream> rateLimiter ) { if (fileLength < 0) { throw new IllegalArgumentException("fileLength must be >= 0, got " + fileLength); } if (partSize <= 0) { throw new IllegalArgumentException("partSize must be > 0, got " + partSize); } - this.blobContainer = blobContainer; - this.blobName = blobName; + this.blobContainer = java.util.Objects.requireNonNull(blobContainer); + this.blobName = java.util.Objects.requireNonNull(blobName); + this.executor = java.util.Objects.requireNonNull(executor); + this.permits = java.util.Objects.requireNonNull(permits); + this.rateLimiter = java.util.Objects.requireNonNull(rateLimiter); this.fileLength = fileLength; this.partSize = partSize; this.numParts = Math.toIntExact((fileLength + partSize - 1) / partSize); Suggestion importance[1-10]: 2 __ Why: The schedulePrefetches method already null-checks permits, but other paths (e.g., Prefetch.releasePermit) would NPE. However, in practice callers always pass non-null values, making this a marginal defensive improvement.	Low

Suggestions up to commit 076b9e2

Category	Suggestion	Impact
Possible issue	Prevent integer overflow computing part offset The expression partIndex * partSize is computed in int arithmetic because partIndex is an int and can overflow for large blobs (e.g. partIndex=2148 with partSize=1MB overflows). Cast to long before multiplication to prevent wrong offsets. The same bug exists in Prefetch#run where partIndex * partSize is used. server/src/main/java/org/opensearch/index/store/ParallelPartInputStream.java [190-191] private void openNextPart() throws IOException { final int partIndex = nextPart++; - final long position = partIndex * partSize; + final long position = (long) partIndex * partSize; final long length = Math.min(partSize, fileLength - position); Suggestion importance[1-10]: 9 __ Why: Correctly identifies a real integer-overflow bug: partIndex * partSize is computed in int arithmetic and can overflow for large blobs, causing prefetch and read to fetch from the wrong offset. The fix (cast to long) is accurate.	High
	Prevent integer overflow in prefetch offset Same integer-overflow issue as openNextPart: partIndex * partSize must be widened to long before multiplication, otherwise prefetch tasks for later parts of very large blobs will read from the wrong offset. server/src/main/java/org/opensearch/index/store/ParallelPartInputStream.java [339-341] try { - final long position = partIndex * partSize; + final long position = (long) partIndex * partSize; final long length = Math.min(partSize, fileLength - position); final byte[] bytes = readRange(position, length); Suggestion importance[1-10]: 9 __ Why: Same overflow bug appears in Prefetch#run where partIndex * partSize is used; widening to long prevents incorrect offsets for large blobs. This is a legitimate correctness fix.	High
General	Fail cancelled prefetch futures on close On close, a prefetch task that has already completed successfully has its future holding a byte[] buffer that is never consumed; simply releasing the permit leaves the buffer reachable via the completed future until prefetches.clear() drops the reference. More importantly, if a prefetch task is currently running (Owner.PREFETCHER) it will still call future.complete(bytes) after close — the buffer becomes garbage immediately, which is fine, but you should also complete the future exceptionally here for any tasks not yet started to make stolen-but-closed reads fail fast rather than block. server/src/main/java/org/opensearch/index/store/ParallelPartInputStream.java [172-176] for (Prefetch prefetch : prefetches.values()) { - prefetch.owner.compareAndSet(Owner.NONE, Owner.CANCELLED); + if (prefetch.owner.compareAndSet(Owner.NONE, Owner.CANCELLED)) { + prefetch.future.completeExceptionally(new IOException("Stream for blob [" + blobName + "] is closed")); + } prefetch.releasePermit(); } prefetches.clear(); Suggestion importance[1-10]: 3 __ Why: Since the stream is single-reader and closed, no one waits on those futures, so failing them fast is a minor robustness improvement rather than a bug fix.	Low

Suggestions up to commit 334a9d7

Category	Suggestion	Impact
Possible issue	Prevent integer overflow when computing part offset The expression partIndex * partSize performs multiplication in int space (since partIndex is int), which will overflow for large files whose part index exceeds ~2^31 / partSize. Cast to long before multiplying to avoid computing a negative or wrapped position. The same issue exists in Prefetch.run(). server/src/main/java/org/opensearch/index/store/ParallelPartInputStream.java [190-191] private void openNextPart() throws IOException { final int partIndex = nextPart++; - final long position = partIndex * partSize; + final long position = (long) partIndex * partSize; final long length = Math.min(partSize, fileLength - position); Suggestion importance[1-10]: 8 __ Why: Correctly identifies that partIndex * partSize is computed in int arithmetic and can overflow for large files, resulting in a negative or wrong offset. Casting to long is a valid and impactful fix.	Medium
	Fix integer overflow computing prefetch offset Same integer overflow bug as in openNextPart: partIndex * partSize is computed in int arithmetic and will wrap for large partIndex. Cast partIndex to long before multiplication. server/src/main/java/org/opensearch/index/store/ParallelPartInputStream.java [329-330] @Override public void run() { if (owner.compareAndSet(Owner.NONE, Owner.PREFETCHER) == false) { // Stolen by the reader or cancelled before we started. releasePermit(); return; } try { - final long position = partIndex * partSize; + final long position = (long) partIndex * partSize; final long length = Math.min(partSize, fileLength - position); Suggestion importance[1-10]: 8 __ Why: Same valid overflow concern applied to Prefetch.run(). Important correctness fix for large files.	Medium
General	Avoid executor contention between copy and prefetch Using the same REMOTE_RECOVERY executor that drives file copies to also run prefetch tasks risks deadlock/starvation: all threads workers are consumed by copyOneFile, each blocking on future.get() for a prefetch task queued behind them on the same pool. Although the reader can steal queued parts, throughput will collapse and, with a bounded queue, RejectedExecutionException becomes the norm. Consider a dedicated executor or ensure the pool size is strictly greater than maxConcurrentRemoteStoreStreams. server/src/main/java/org/opensearch/index/store/RemoteStoreFileDownloader.java [155-166] return new FilterDirectory(source) { @Override public IndexInput openInput(String name, IOContext context) throws IOException { final long fileLength = in.fileLength(name); if (fileLength <= partSize) { return in.openInput(name, context); } logger.trace("Downloading file {} of size {} in parts of {} bytes", name, fileLength, partSize); + // TODO: use a dedicated executor to avoid contending with copyOneFile workers on REMOTE_RECOVERY return remoteSource.openParallelInput(name, fileLength, context, partSize, executor, permits); } }; Suggestion importance[1-10]: 6 __ Why: Raises a legitimate concern about executor contention/potential starvation when the same pool runs both copy workers and prefetch tasks. The stream is designed to fall back gracefully (reader steals queued parts, rejection handled), but the observation is worth investigating.	Low
	Reset stream state on prefetch failure When the prefetch task fails, current is never assigned but currentRemaining is later set to length and currentPrefetch remains unset from a prior part. A subsequent read() call after catching the exception (or the close() path) may misbehave. Also, on EOFException mid-part, nextPart has already been incremented, so the error message references nextPart - 1 correctly, but state is inconsistent. Ensure current/currentRemaining are safely reset (e.g., set currentRemaining = 0) before throwing so that state is consistent. server/src/main/java/org/opensearch/index/store/ParallelPartInputStream.java [214-217] } catch (ExecutionException e) { prefetch.releasePermit(); + currentRemaining = 0; + current = null; + currentPrefetch = null; throw unwrap(e.getCause(), partIndex); } Suggestion importance[1-10]: 4 __ Why: The concern about inconsistent state after a prefetch failure has some merit, but currentRemaining isn't yet set at that point (it's set after the try/catch), so the suggested reset is partially misplaced. Minor defensive improvement.	Low

[github-actions]

✅ Gradle check result for 334a9d7: SUCCESS

[codecov]

Codecov Report

❌ Patch coverage is 92.50000% with 3 lines in your changes missing coverage. Please review.
✅ Project coverage is 71.75%. Comparing base (b62087d) to head (89bc4fc).

Files with missing lines	Patch %	Lines
...earch/index/store/RemoteSegmentStoreDirectory.java	50.00%	1 Missing and 1 partial ⚠️
...x/store/remote/DataFormatAwareRemoteDirectory.java	0.00%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main   #23016      +/-   ##
============================================
+ Coverage     71.66%   71.75%   +0.08%     
- Complexity    77565    77658      +93     
============================================
  Files          6168     6168              
  Lines        360105   360144      +39     
  Branches      52380    52383       +3     
============================================
+ Hits         258086   258425     +339     
+ Misses        81434    81210     -224     
+ Partials      20585    20509      -76     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[github-actions]

Persistent review updated to latest commit 076b9e2

[github-actions]

✅ Gradle check result for 076b9e2: SUCCESS

[github-actions]

Persistent review updated to latest commit 00bcf7f

[github-actions]

Persistent review updated to latest commit 89bc4fc

[github-actions]

✅ Gradle check result for 89bc4fc: SUCCESS