Fix bulk doc_as_upsert ingest pipeline to run after merge

[habiibullahm]

Summary

• Fixes [BUG] Bulk upsert does not behave like a single Upsert, with an ingestion pipeline #10864: bulk doc_as_upsert with an ingest pipeline now matches single _update behavior by running pipelines on the full document after UpdateHelper.prepare (create or merged update), instead of on the partial doc at the coordinating node.
• Coordinating node marks doc_as_upsert child docs as NOOP so pre-merge ingest is skipped; TransportShardBulkAction re-resolves and executes pipelines on the primary after prepare.

Test plan

• TransportBulkActionIngestTests / TransportShardBulkActionTests unit tests
• YAML regression in ingest/70_bulk.yml reproducing the issue scenario (duration/begin/end pipeline on existing doc)
• CI gradle-check on this PR

Made with Cursor

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit 78d1d4e)

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 Double listener invocation In executeIngestThenApply, after the ingest completion callback runs applyRequestToExecute, if it returns true the code calls itemDoneListener.onResponse(null). However, applyRequestToExecute may itself invoke the listener via completion paths (e.g. onComplete -> context.markAsCompleted/failure paths inside executeOnPrimaryWhileHandlingMappingUpdates-like flow in the original method). The pre-refactor code returned true to indicate the caller's outer loop advances; the listener was invoked by the outer ActionRunnable, not by executeBulkItemRequest. Now the ingest-then-apply path invokes itemDoneListener.onResponse(null) directly when completed==true, which may double-invoke the listener and cause the outer while loop to advance twice or trigger assertions. Please verify the listener contract carefully. final Runnable apply = () -> { try { boolean completed = applyRequestToExecute( context, updateResult, mappingUpdater, waitForMappingUpdate, itemDoneListener ); if (completed) { itemDoneListener.onResponse(null); } } catch (Exception e) { itemDoneListener.onFailure(e); } }; if (originalThread == Thread.currentThread()) { apply.run(); } else { context.getPrimary().getThreadPool().executor(executorName).execute(apply); } }, Possible NPE on drop In the drop handler for executeBulkRequest, context.markOperationAsNoOp is called followed by context.markAsCompleted(context.getExecutionResult()), but the context state was set to TRANSLATED via setRequestToExecute prior to calling ingest. Ensure the context state machine permits markOperationAsNoOp from TRANSLATED for update items; otherwise dropped documents will trigger an IllegalStateException. Also noopUpdate() is invoked directly on the primary shard while any pending refresh/replication accounting for the original translated request may be inconsistent. slot -> { // Document dropped by pipeline — treat as noop for this bulk item context.getPrimary().noopUpdate(); context.markOperationAsNoOp( new UpdateResponse( context.getPrimary().shardId(), indexRequest.id(), SequenceNumbers.UNASSIGNED_SEQ_NO, SequenceNumbers.UNASSIGNED_PRIMARY_TERM, indexRequest.version(), DocWriteResponse.Result.NOOP ) ); context.markAsCompleted(context.getExecutionResult()); itemDoneListener.onResponse(null); }, Pipeline resolution cache bypass The NOOP markers are cleared only when both getPipeline and getFinalPipeline are NOOP. If a user request happens to legitimately have NOOP for one but a real pipeline for the other after coordinating resolution (unlikely today but possible), the re-resolution on the primary will be skipped and the deferred pipelines won't execute. Consider a more explicit flag on the request (e.g., a "deferred for docAsUpsert" marker) rather than relying on the NOOP sentinel pattern, which conflates "no pipeline" with "deferred". // Clear coordinating-node NOOP markers when reusing the doc IndexRequest (create path) if (indexRequest.isPipelineResolved() && IngestService.NOOP_PIPELINE_NAME.equals(indexRequest.getPipeline()) && IngestService.NOOP_PIPELINE_NAME.equals(indexRequest.getFinalPipeline())) { indexRequest.isPipelineResolved(false); indexRequest.setPipeline(null); indexRequest.setFinalPipeline(null); indexRequest.setSystemIngestPipeline(null); } Backward-compat overload defaults The new 9-arg performOnPrimary and 6-arg executeBulkItemRequest overloads pass null for ingestService and Metadata.EMPTY_METADATA supplier. Any external caller (or test) using these overloads will silently skip the doc_as_upsert ingest fix. This may hide regressions since ingestService != null is required to trigger the new path. Consider making the ingest-aware path mandatory or logging when it is skipped. ) { performOnPrimary( request, primary, updateHelper, nowInMillisSupplier, mappingUpdater, waitForMappingUpdate, listener, threadPool, executorName, null, () -> Metadata.EMPTY_METADATA ); }

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 78d1d4e
Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Avoid double-invoking the item listener applyRequestToExecute already invokes itemDoneListener via its internal completion paths (e.g., markAsCompleted/onComplete for successful/failed writes). Explicitly calling itemDoneListener.onResponse(null) again when it returns true will double-invoke the listener, which can corrupt bulk item accounting or trigger executor re-submission twice. Remove the extra onResponse call and rely on the same contract used by the non-ingest path. server/src/main/java/org/opensearch/action/bulk/TransportShardBulkAction.java [764-779] final Runnable apply = () -> { try { - boolean completed = applyRequestToExecute( + applyRequestToExecute( context, updateResult, mappingUpdater, waitForMappingUpdate, itemDoneListener ); - if (completed) { - itemDoneListener.onResponse(null); - } } catch (Exception e) { itemDoneListener.onFailure(e); } }; Suggestion importance[1-10]: 7 __ Why: The concern is legitimate — applyRequestToExecute may internally trigger completion paths that already signal the listener (via markAsCompleted/onComplete), so unconditionally invoking itemDoneListener.onResponse(null) when it returns true could cause double-invocation. However, verifying the exact contract requires broader code inspection.	Medium
	Prevent duplicate listener signaling on ingest failure After ingest failure, onComplete typically transitions the context to completed and may itself signal item completion; unconditionally calling itemDoneListener.onResponse(null) afterwards risks double-signal, and the outer loop in performOnPrimary re-executes on the same thread which can also re-enter. Ensure the listener is invoked exactly once by checking whether the context has already been marked completed, or by mirroring the exact contract of the non-ingest path. server/src/main/java/org/opensearch/action/bulk/TransportShardBulkAction.java [755-763] (slot, e) -> itemFailure.set(e), (originalThread, exception) -> { final Exception failure = exception != null ? exception : itemFailure.get(); if (failure != null) { final Engine.Result result = exceptionToResult(failure, context.getPrimary(), false, indexRequest.version()); onComplete(result, context, updateResult); + // onComplete already advances the context; signal item done exactly once itemDoneListener.onResponse(null); return; } Suggestion importance[1-10]: 3 __ Why: The suggestion raises a concern but the improved_code is essentially identical to the existing_code (just adds a comment), providing no actual fix. Low impact.	Low
General	Include system pipeline in NOOP marker check The reset condition only checks default and final pipelines, ignoring systemIngestPipeline. If the coordinating node set the system pipeline to NOOP but a real system pipeline should be resolved after prepare, this check may still pass; conversely, if the system pipeline was set to a non-NOOP value elsewhere, the reset should still occur for correctness. Include getSystemIngestPipeline() in the equality check to make the "coordinating-node NOOP marker" detection robust. server/src/main/java/org/opensearch/action/bulk/TransportShardBulkAction.java [698-705] if (indexRequest.isPipelineResolved() && IngestService.NOOP_PIPELINE_NAME.equals(indexRequest.getPipeline()) - && IngestService.NOOP_PIPELINE_NAME.equals(indexRequest.getFinalPipeline())) { + && IngestService.NOOP_PIPELINE_NAME.equals(indexRequest.getFinalPipeline()) + && IngestService.NOOP_PIPELINE_NAME.equals(indexRequest.getSystemIngestPipeline())) { indexRequest.isPipelineResolved(false); indexRequest.setPipeline(null); indexRequest.setFinalPipeline(null); indexRequest.setSystemIngestPipeline(null); } Suggestion importance[1-10]: 5 __ Why: Adding the system pipeline to the NOOP marker check makes detection more robust and consistent with how the coordinating node sets all three fields to NOOP, reducing edge cases.	Low
	Guard against double completion of item The drop callback signals completion via itemDoneListener.onResponse(null), but performOnPrimary wraps this listener to re-execute the runnable on the executor. If the completion handler is invoked before the drop handler (or both fire), the item could be advanced twice. Ensure the ingest service contract guarantees that either the drop handler or the completion handler is called for a given slot, and add an assertion or guard (e.g., an AtomicBoolean invoked-once flag) to protect against accidental double-invocation. server/src/main/java/org/opensearch/action/bulk/TransportShardBulkAction.java [786-801] +final java.util.concurrent.atomic.AtomicBoolean signaled = new java.util.concurrent.atomic.AtomicBoolean(false); +// ... slot -> { - // Document dropped by pipeline — treat as noop for this bulk item + if (signaled.compareAndSet(false, true) == false) return; context.getPrimary().noopUpdate(); context.markOperationAsNoOp( new UpdateResponse( context.getPrimary().shardId(), indexRequest.id(), SequenceNumbers.UNASSIGNED_SEQ_NO, SequenceNumbers.UNASSIGNED_PRIMARY_TERM, indexRequest.version(), DocWriteResponse.Result.NOOP ) ); context.markAsCompleted(context.getExecutionResult()); itemDoneListener.onResponse(null); }, Suggestion importance[1-10]: 4 __ Why: Defensive guarding against double-invocation is reasonable, but the ingest service contract typically ensures either drop or completion is called, not both. The suggestion is speculative without evidence of actual double-invocation risk.	Low

---

Previous suggestions

Suggestions up to commit 42c49fc

Category	Suggestion	Impact
Possible issue	Avoid double-invoking the item listener The itemDoneListener may be invoked twice on success: once here after applyRequestToExecute returns true, and again by applyRequestToExecute itself via its inner listeners (which call itemDoneListener.onResponse when the item completes). Rely on applyRequestToExecute's existing listener contract and do not additionally invoke itemDoneListener.onResponse(null) from apply. server/src/main/java/org/opensearch/action/bulk/TransportShardBulkAction.java [760-775] ingestService.executeBulkRequest(1, Collections.singletonList(indexRequest), (slot, e) -> itemFailure.set(e), (originalThread, exception) -> { final Exception failure = exception != null ? exception : itemFailure.get(); if (failure != null) { final Engine.Result result = exceptionToResult(failure, context.getPrimary(), false, indexRequest.version()); onComplete(result, context, updateResult); itemDoneListener.onResponse(null); return; } final Runnable apply = () -> { try { - boolean completed = applyRequestToExecute( - context, - updateResult, - mappingUpdater, - waitForMappingUpdate, - itemDoneListener - ); - if (completed) { - itemDoneListener.onResponse(null); - } + applyRequestToExecute(context, updateResult, mappingUpdater, waitForMappingUpdate, itemDoneListener); } catch (Exception e) { itemDoneListener.onFailure(e); } }; Suggestion importance[1-10]: 8 __ Why: Valid concern: applyRequestToExecute completes items via its own inner listeners that call itemDoneListener.onResponse, and the wrapping code also invokes itemDoneListener.onResponse(null) when completed is true, likely causing double invocation and potential incorrect bulk accounting.	Medium
General	Dispatch dropped-doc completion on write executor The dropped-document callback is invoked from the ingest thread; calling context.markAsCompleted and itemDoneListener.onResponse(null) inline may execute subsequent bulk items on the ingest thread rather than the write executor. Dispatch the noop-completion path via context.getPrimary().getThreadPool().executor(executorName) for consistency with the success path. server/src/main/java/org/opensearch/action/bulk/TransportShardBulkAction.java [781-796] }, slot -> { - // Document dropped by pipeline — treat as noop for this bulk item - context.getPrimary().noopUpdate(); - context.markOperationAsNoOp( - new UpdateResponse( - context.getPrimary().shardId(), - indexRequest.id(), - SequenceNumbers.UNASSIGNED_SEQ_NO, - SequenceNumbers.UNASSIGNED_PRIMARY_TERM, - indexRequest.version(), - DocWriteResponse.Result.NOOP - ) - ); - context.markAsCompleted(context.getExecutionResult()); - itemDoneListener.onResponse(null); + context.getPrimary().getThreadPool().executor(executorName).execute(() -> { + context.getPrimary().noopUpdate(); + context.markOperationAsNoOp( + new UpdateResponse( + context.getPrimary().shardId(), + indexRequest.id(), + SequenceNumbers.UNASSIGNED_SEQ_NO, + SequenceNumbers.UNASSIGNED_PRIMARY_TERM, + indexRequest.version(), + DocWriteResponse.Result.NOOP + ) + ); + context.markAsCompleted(context.getExecutionResult()); + itemDoneListener.onResponse(null); + }); }, executorName); Suggestion importance[1-10]: 5 __ Why: Reasonable consistency improvement to avoid running subsequent bulk items on the ingest thread, though the impact depends on downstream behavior and may not be strictly required.	Low
	Verify NOOP marker does not skip primary re-resolve Because this branch now returns false from indexRequestHasPipeline for docAsUpsert, callers that determine whether to run ingest on the coordinator based on the pipeline flag may skip the upsert's pipeline entirely if only the doc was inspected. Ensure the return value still reflects that the upsertRequest above has a pipeline (already handled via |= for upsert), but verify that downstream logic does not treat this NOOP marker as "pipeline resolved with real pipeline" causing the primary-side re-resolution to be skipped when isPipelineResolved() is already true. server/src/main/java/org/opensearch/action/bulk/TransportBulkAction.java [434-439] +if (docRequest.isPipelineResolved() == false) { + docRequest.setPipeline(IngestService.NOOP_PIPELINE_NAME); + docRequest.setFinalPipeline(IngestService.NOOP_PIPELINE_NAME); + docRequest.setSystemIngestPipeline(IngestService.NOOP_PIPELINE_NAME); + docRequest.isPipelineResolved(true); +} - Suggestion importance[1-10]: 2 __ Why: The suggestion only asks to verify behavior and provides identical existing_code and improved_code, offering no actionable change.	Low

[github-actions]

❌ Gradle check result for 42c49fc: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?

[habiibullahm]

Root cause of the failed gradle-check on 42c49fc: Spotless formatting violations in the touched Java files (precommit fails early — not a test logic regression).

Pushed a formatting-only follow-up with DCO sign-off: 78d1d4e5841. Waiting on a fresh gradle-check.

[github-actions]

Persistent review updated to latest commit 78d1d4e

[github-actions]

❌ Gradle check result for 78d1d4e: FAILURE

Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change?