Make BulkItemRequest immutable by msfroh · Pull Request #20831 · opensearch-project/OpenSearch

msfroh · 2026-03-10T23:12:20Z

Description

I was talking with @itschrispeck about some JIT optimization issues in BulkItemRequest's serialization. While looking at the code, the volatile keyword on the primaryResponse field made me cringe. Why is a BulkItemRequest mutable at all?

It turns out that we modify the existing BulkItemRequest instances on the primary shard. These modified requests (as part of a BulkShardRequest) are sent to the replicas. That is, the PrimaryExecutionContext returns the modified-in-place BulkShardRequest that it was created with, which is sent to the replicas.

This change makes BulkItemRequest immutable. The PrimaryExecutionContext collects all of the primary responses, then produces a new BulkShardRequest that combines the original BulkItemRequests with the responses, which gets forwarded to replicas.

Related Issues

N/A

Check List

Functionality includes testing.
API changes companion pull request created, if applicable.
Public documentation issue/PR created, if applicable.

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 · 2026-03-10T23:13:47Z

PR Reviewer Guide 🔍

(Review updated until commit `d2a6232`)

Here are some key observations to aid the review process:

🧪 PR contains tests
🔒 No security concerns identified
✅ No TODO sections
🔀 Multiple PR themes Sub-PR theme: Convert BulkItemRequest to immutable record Relevant files: server/src/main/java/org/opensearch/action/bulk/BulkItemRequest.java server/src/main/java/org/opensearch/action/bulk/BulkItemResponse.java Sub-PR theme: Update execution context and action to use immutable BulkItemRequest Relevant files: server/src/main/java/org/opensearch/action/bulk/BulkPrimaryExecutionContext.java server/src/main/java/org/opensearch/action/bulk/TransportShardBulkAction.java server/src/test/java/org/opensearch/action/bulk/BulkPrimaryExecutionContextTests.java server/src/test/java/org/opensearch/action/bulk/TransportShardBulkActionTests.java
⚡ Recommended focus areas for review Null Dereference In `performOnReplica`, `item.primaryResponse()` is called twice: once to assign to `response`, and again immediately to call `.isFailed()`. If `primaryResponse()` returns null (which is valid per the record definition — the constructor accepting only `id` and `request` sets it to null), the second call `item.primaryResponse().isFailed()` will throw a NullPointerException. The old code used `item.getPrimaryResponse()` in the same pattern, but the new immutable design makes null a more likely scenario since items forwarded to replicas without a primary response would crash here. final BulkItemResponse response = item.primaryResponse(); final Engine.Result operationResult; if (item.primaryResponse().isFailed()) { Incomplete Abort Handling The `abort` method was removed from `BulkItemRequest` as part of making it immutable, but the `findNextNonAborted` method in `BulkPrimaryExecutionContext` now checks `request.items()[startIndex].primaryResponse()` to detect aborted items. Since the primary request items are constructed without a `primaryResponse` (null), pre-aborted items can no longer be skipped. The tests `testAbortedSkipped` and `testSkipBulkIndexRequestIfAborted` were deleted rather than adapted, removing coverage for this behavior. If callers previously relied on pre-aborting items before passing to `BulkPrimaryExecutionContext`, those items will now be processed instead of skipped. while (startIndex < length && isAborted(request.items()[startIndex].primaryResponse())) { startIndex++; } Missing Properties in Clone `getBulkShardRequest()` copies `waitForActiveShards`, `timeout`, `routedBasedOnClusterVersion`, and `parentTask` from the original request, but may miss other mutable properties set on `BulkShardRequest` (e.g., `canReturnNullResponseIfNoShardAvailable`, or any future fields). This is fragile — if `BulkShardRequest` gains new properties, the replica request will silently miss them. BulkShardRequest bulkShardRequest = new BulkShardRequest(request.shardId(), request.getRefreshPolicy(), newRequests); // Clone other properties from primary shard request // See TransportBulkAction.BulkOperation#doRun() for construction of the primary shard request. bulkShardRequest.waitForActiveShards(request.waitForActiveShards()); bulkShardRequest.timeout(request.timeout()); bulkShardRequest.routedBasedOnClusterVersion(request.routedBasedOnClusterVersion()); bulkShardRequest.setParentTask(request.getParentTask()); return bulkShardRequest;

github-actions · 2026-03-10T23:14:17Z

PR Code Suggestions ✨

Latest suggestions up to d2a6232

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Fix abort-skipping logic to use new responses array After making `BulkItemRequest` immutable, aborted items can no longer be pre-marked via `setPrimaryResponse`. The `findNextNonAborted` method now checks `primaryResponse()` on the original request items, but aborted responses are stored in `primaryResponses[]`. This means aborted items will never be skipped, breaking the abort-skipping logic entirely. server/src/main/java/org/opensearch/action/bulk/BulkPrimaryExecutionContext.java [105] -while (startIndex < length && isAborted(request.items()[startIndex].primaryResponse())) { +while (startIndex < length && isAborted(primaryResponses[startIndex])) { Suggestion importance[1-10]: 9 __ Why: This is a critical correctness issue: since `BulkItemRequest` is now immutable, `request.items()[startIndex].primaryResponse()` will always return the original value (never an aborted response), so the abort-skipping logic is completely broken. The fix to use `primaryResponses[startIndex]` is correct and necessary.	High
Possible issue	Fix incorrect assertion on immutable record field The `COMPLETED` state assertion now checks `getCurrentItem().primaryResponse()` on the original (immutable) `BulkItemRequest`, but primary responses are now stored in the separate `primaryResponses` array rather than in the item itself. This assertion will always fail since the record's `primaryResponse` field is never mutated. It should check `primaryResponses[currentIndex]` instead. server/src/main/java/org/opensearch/action/bulk/BulkPrimaryExecutionContext.java [390] -assert getCurrentItem().primaryResponse() != null; +assert primaryResponses[currentIndex] != null; Suggestion importance[1-10]: 8 __ Why: This is a valid bug: `BulkItemRequest` is now an immutable record, so `getCurrentItem().primaryResponse()` will always return the original value (likely `null`) rather than the one stored in `primaryResponses[currentIndex]`. The assertion would incorrectly fail or pass depending on the original record state.	Medium
General	Guard against premature bulk shard request construction `getBulkShardRequest()` is called multiple times in tests and potentially in production code paths. Each call creates a new `BulkShardRequest` with new `BulkItemRequest` instances, which is expensive and may cause inconsistencies if called before all items are completed. Consider computing this once lazily or only when `hasMoreOperationsToExecute()` is false. server/src/main/java/org/opensearch/action/bulk/BulkPrimaryExecutionContext.java [133-145] public BulkShardRequest getBulkShardRequest() { + assert hasMoreOperationsToExecute() == false : "getBulkShardRequest() called before all operations are completed"; BulkItemRequest[] newRequests = new BulkItemRequest[request.items().length]; for (int i = 0; i < newRequests.length; i++) { BulkItemRequest oldRequest = request.items()[i]; newRequests[i] = new BulkItemRequest(oldRequest.id(), oldRequest.request(), primaryResponses[i]); } BulkShardRequest bulkShardRequest = new BulkShardRequest(request.shardId(), request.getRefreshPolicy(), newRequests); - ... + bulkShardRequest.waitForActiveShards(request.waitForActiveShards()); + bulkShardRequest.timeout(request.timeout()); + bulkShardRequest.routedBasedOnClusterVersion(request.routedBasedOnClusterVersion()); + bulkShardRequest.setParentTask(request.getParentTask()); return bulkShardRequest; } Suggestion importance[1-10]: 5 __ Why: Adding an assertion to guard against calling `getBulkShardRequest()` before all operations are completed is a reasonable defensive measure, but it's a minor improvement since `buildShardResponse` already has such an assertion and the method is primarily called after completion in tests.	Low

Previous suggestions

Suggestions up to commit 93814fc

Category	Suggestion	Impact
General	Avoid mutating original request items array When `markAsCompleted` updates `request.items()[currentIndex]` with a new `BulkItemRequest` (for translated requests like updates), the `getBulkShardRequest()` method later reads `oldRequest.request()` from `request.items()[i]`. This means the translated `requestToExecute` is correctly captured in the items array. However, `primaryResponses[currentIndex]` is stored separately and then combined in `getBulkShardRequest()`. This is consistent, but the update to `request.items()` is still mutating the original `BulkShardRequest`'s items array, which undermines the immutability goal. Consider storing translated requests in a separate array instead of mutating `request.items()`. server/src/main/java/org/opensearch/action/bulk/BulkPrimaryExecutionContext.java [342-345] +// In BulkPrimaryExecutionContext, add a field: +private final DocWriteRequest<?>[] translatedRequests; + +// In constructor: +this.translatedRequests = new DocWriteRequest<?>[request.items().length]; + +// In markAsCompleted: if (translatedResponse.isFailed() == false && requestToExecute != null && requestToExecute != getCurrent()) { - request.items()[currentIndex] = new BulkItemRequest(request.items()[currentIndex].id(), requestToExecute); + translatedRequests[currentIndex] = requestToExecute; } primaryResponses[currentIndex] = translatedResponse; +// In getBulkShardRequest: +BulkItemRequest oldRequest = request.items()[i]; +DocWriteRequest<?> effectiveRequest = translatedRequests[i] != null ? translatedRequests[i] : oldRequest.request(); +newRequests[i] = new BulkItemRequest(oldRequest.id(), effectiveRequest, primaryResponses[i]); + Suggestion importance[1-10]: 5 __ Why: The suggestion correctly identifies that `request.items()[currentIndex]` is still being mutated in `markAsCompleted`, which partially undermines the immutability goal of the PR. However, the `improved_code` is a multi-part snippet spanning multiple methods and fields, making it harder to evaluate precisely. The concern is valid but the implementation is complex and the `getBulkShardRequest()` method already reads from `request.items()` to get the (potentially mutated) request, so the current approach is at least consistent.	Low
General	Guard against premature bulk request snapshot The `getBulkShardRequest()` method creates a new `BulkShardRequest` on every call, which is expensive and may produce inconsistent snapshots if called multiple times during processing. Since this method is called both during processing (to update item requests) and at the end to retrieve results, consider only building the new request once after all operations are complete, or caching the result after processing is done. server/src/main/java/org/opensearch/action/bulk/BulkPrimaryExecutionContext.java [133-139] public BulkShardRequest getBulkShardRequest() { + assert hasMoreOperationsToExecute() == false : "getBulkShardRequest() called before all operations are complete"; BulkItemRequest[] newRequests = new BulkItemRequest[request.items().length]; for (int i = 0; i < newRequests.length; i++) { BulkItemRequest oldRequest = request.items()[i]; newRequests[i] = new BulkItemRequest(oldRequest.id(), oldRequest.request(), primaryResponses[i]); } return new BulkShardRequest(request.shardId(), request.getRefreshPolicy(), newRequests); } Suggestion importance[1-10]: 4 __ Why: Adding an assertion to prevent `getBulkShardRequest()` from being called before all operations complete is a reasonable defensive check. However, looking at the test code, `getBulkShardRequest()` is called after `assertFalse(context.hasMoreOperationsToExecute())`, so this is more of a safety guard than fixing a real bug. The improvement is minor.	Low
Possible issue	Verify request identity preserved after immutability refactor The loop iterates over `completedRequest[0].items()` but calls `verify` with `eq(updateRequest)` where `updateRequest` is cast from `item.request()`. Since `getBulkShardRequest()` now creates new `BulkItemRequest` objects, the `updateRequest` reference from the completed request may differ from the original `UpdateRequest` used in `when(updateHelper.prepare(...))`. Verify that the `UpdateRequest` identity is preserved through the new `BulkItemRequest` construction to ensure Mockito's `eq()` matcher works correctly. server/src/test/java/org/opensearch/action/bulk/TransportShardBulkActionTests.java [1232-1243] for (BulkItemRequest item : completedRequest[0].items()) { + assertNotNull(item.getPrimaryResponse()); assertEquals(item.getPrimaryResponse().getFailure().getCause().getClass(), VersionConflictEngineException.class); - // this assertion is based on the assumption that all bulk item requests are updates and are hence calling - // UpdateRequest::prepareRequest UpdateRequest updateRequest = (UpdateRequest) item.request(); + // Ensure the same UpdateRequest instance is used for verification + assertSame(updateRequest, items.stream() + .filter(orig -> orig.id() == item.id()) + .findFirst().get().request()); verify(updateHelper, times(updateRequest.retryOnConflict() + 1)).prepare( eq(updateRequest), any(IndexShard.class), any(LongSupplier.class) ); } Suggestion importance[1-10]: 5 __ Why: The concern about `UpdateRequest` identity through the new `BulkItemRequest` construction is valid - since `getBulkShardRequest()` creates new `BulkItemRequest` objects wrapping the same `DocWriteRequest` references, the `eq()` matcher should still work as object identity is preserved. However, the `improved_code` adds unnecessary complexity with a stream lookup. The suggestion raises a legitimate concern but the actual risk is low since the request references are passed through unchanged.	Low

Suggestions up to commit 1201d59

Category	Suggestion	Impact
General	Guard against incomplete state in request builder This method is called to get the bulk shard request, but it creates a new `BulkShardRequest` on every invocation, which could be expensive if called multiple times. More critically, `primaryResponses[i]` may be `null` for items that haven't been processed yet, which could cause issues downstream if callers expect all responses to be populated. Consider documenting this behavior or asserting that all responses are set before calling this method. server/src/main/java/org/opensearch/action/bulk/BulkPrimaryExecutionContext.java [132-139] public BulkShardRequest getBulkShardRequest() { + assert hasMoreOperationsToExecute() == false : "getBulkShardRequest called before all operations are completed"; BulkItemRequest[] newRequests = new BulkItemRequest[request.items().length]; for (int i = 0; i < newRequests.length; i++) { BulkItemRequest oldRequest = request.items()[i]; newRequests[i] = new BulkItemRequest(oldRequest.id(), oldRequest.request(), primaryResponses[i]); } return new BulkShardRequest(request.shardId(), request.getRefreshPolicy(), newRequests); } Suggestion importance[1-10]: 5 __ Why: Adding an assertion that `hasMoreOperationsToExecute() == false` before building the request is a valid defensive check, as `primaryResponses[i]` could be `null` for unprocessed items. This mirrors the pattern already used in `buildShardResponse`.	Low
	Defensive copy prevents internal array mutation The `primaryResponses` array is passed directly to `BulkShardResponse`, which may allow external mutation of the internal state. A defensive copy should be used to prevent unintended modifications to the array after the response is built. server/src/main/java/org/opensearch/action/bulk/BulkPrimaryExecutionContext.java [351-354] public BulkShardResponse buildShardResponse(long serviceTimeEWMAInNanos, int nodeQueueSize) { assert hasMoreOperationsToExecute() == false; - return new BulkShardResponse(request.shardId(), primaryResponses, serviceTimeEWMAInNanos, nodeQueueSize); + return new BulkShardResponse(request.shardId(), Arrays.copyOf(primaryResponses, primaryResponses.length), serviceTimeEWMAInNanos, nodeQueueSize); } Suggestion importance[1-10]: 3 __ Why: While defensive copying is a good practice, `primaryResponses` is a private final field only accessible within this class, and `BulkShardResponse` likely treats it as immutable. The risk of external mutation is low in this context.	Low
	Verify response array alignment with item array When a translated request (e.g., update translated to index) replaces the original item in `request.items()`, the `primaryResponses` array is indexed by `currentIndex` which aligns with the original `request.items()` array. However, `getBulkShardRequest()` iterates over `request.items()` and pairs each with `primaryResponses[i]`, so the replacement of `request.items()[currentIndex]` with a new `BulkItemRequest` (without a response) and then separately storing the response in `primaryResponses[currentIndex]` is correct. This is fine, but worth verifying that `getBulkShardRequest()` is only called after all operations complete to avoid mismatched null responses. server/src/main/java/org/opensearch/action/bulk/BulkPrimaryExecutionContext.java [342-345] +if (translatedResponse.isFailed() == false && requestToExecute != null && requestToExecute != getCurrent()) { + request.items()[currentIndex] = new BulkItemRequest(request.items()[currentIndex].id(), requestToExecute); +} +primaryResponses[currentIndex] = translatedResponse; - Suggestion importance[1-10]: 1 __ Why: The `existing_code` and `improved_code` are identical, making this suggestion a verification request rather than an actual code change. It offers no actionable improvement to the PR.	Low

Suggestions up to commit 0450fb9

Category	Suggestion	Impact
General	Guard against incomplete state in bulk request retrieval `getBulkShardRequest()` creates a new `BulkShardRequest` on every call, which is expensive if called multiple times. Additionally, `primaryResponses[i]` may be `null` for items that haven't been processed yet, which could cause issues downstream if callers expect all responses to be populated. Consider documenting this behavior or asserting that all responses are set before calling this method. server/src/main/java/org/opensearch/action/bulk/BulkPrimaryExecutionContext.java [132-139] public BulkShardRequest getBulkShardRequest() { + assert hasMoreOperationsToExecute() == false : "getBulkShardRequest called before all items were processed"; BulkItemRequest[] newRequests = new BulkItemRequest[request.items().length]; for (int i = 0; i < newRequests.length; i++) { BulkItemRequest oldRequest = request.items()[i]; newRequests[i] = new BulkItemRequest(oldRequest.id(), oldRequest.request(), primaryResponses[i]); } return new BulkShardRequest(request.shardId(), request.getRefreshPolicy(), newRequests); } Suggestion importance[1-10]: 5 __ Why: Adding an assertion that `hasMoreOperationsToExecute() == false` before building the request is a valid defensive check, similar to how `buildShardResponse` already does this. It prevents misuse when `primaryResponses` may have null entries for unprocessed items.	Low
General	Defensive copy prevents internal array mutation The `primaryResponses` array is passed directly to `BulkShardResponse`, which may allow external mutation of the internal state. A defensive copy should be used to prevent the caller from modifying the array after the response is built. server/src/main/java/org/opensearch/action/bulk/BulkPrimaryExecutionContext.java [351-354] public BulkShardResponse buildShardResponse(long serviceTimeEWMAInNanos, int nodeQueueSize) { assert hasMoreOperationsToExecute() == false; - return new BulkShardResponse(request.shardId(), primaryResponses, serviceTimeEWMAInNanos, nodeQueueSize); + return new BulkShardResponse(request.shardId(), Arrays.copyOf(primaryResponses, primaryResponses.length), serviceTimeEWMAInNanos, nodeQueueSize); } Suggestion importance[1-10]: 3 __ Why: While defensive copying is a good practice, `primaryResponses` is a private final field and `BulkShardResponse` likely stores it internally. The risk of external mutation is low in this context, making this a minor improvement.	Low

github-actions · 2026-03-10T23:54:27Z

❌ Gradle check result for 0450fb9: FAILURE