Implement dual-write by mirror indexing requests at coordinator code

[gaobinlong]

Description

Resolves #22434

Related Issues

#22434

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]

PR Reviewer Guide 🔍

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: Introduce dual-write index settings Relevant files: server/src/main/java/org/opensearch/index/IndexSettings.java server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java Sub-PR theme: Make UpdateHelper.prepare public and relax single-item bulk assertion Relevant files: server/src/main/java/org/opensearch/action/update/UpdateHelper.java server/src/main/java/org/opensearch/action/bulk/TransportSingleItemBulkWriteAction.java Sub-PR theme: Implement dual-write mirror expansion in TransportBulkAction Relevant files: server/src/main/java/org/opensearch/action/bulk/BulkRequest.java server/src/main/java/org/opensearch/action/bulk/TransportBulkAction.java server/src/test/java/org/opensearch/action/bulk/TransportBulkActionIndicesThatCannotBeCreatedTests.java server/src/test/java/org/opensearch/action/bulk/TransportBulkActionIngestTests.java server/src/test/java/org/opensearch/action/bulk/TransportBulkActionTests.java server/src/test/java/org/opensearch/action/bulk/TransportBulkActionTookTests.java server/src/test/java/org/opensearch/snapshots/SnapshotResiliencyTests.java rest-api-spec/src/main/resources/rest-api-spec/test/bulk/120_dual_write.yml

⚡ Recommended focus areas for review Response ordering / index mismatch expandDualWriteRequests fires GETs concurrently and appends mirrored requests to bulkRequest.requests in non-deterministic order via a ConcurrentLinkedQueue. Because BulkResponse items are correlated with the original request slots by index, appending mirrors in an arbitrary order (interleaved with bulkRequest.requests.set(slot, ...) for updates) will produce a response array whose mirror items sit at unpredictable positions. Downstream consumers (and the test at items.1) rely on the mirror being at a specific index, which will be flaky under real concurrency. private void expandDualWriteRequests(BulkRequest bulkRequest, Metadata metadata, ActionListener<Void> listener) { final int originalSize = bulkRequest.requests.size(); final List<DocWriteRequest<?>> pendingRequests = new ArrayList<>(); final List<String> pendingTargetIndices = new ArrayList<>(); final List<Integer> pendingSlots = new ArrayList<>(); for (int i = 0; i < originalSize; i++) { final DocWriteRequest<?> request = bulkRequest.requests.get(i); if (request == null) { continue; } final IndexMetadata indexMetadata = metadata.index(request.index()); if (indexMetadata == null) { continue; } final String dualWriteTarget = IndexSettings.INDEX_DUAL_WRITE_INDEX_NAME_SETTING.get(indexMetadata.getSettings()); if (dualWriteTarget == null || dualWriteTarget.isEmpty()) { continue; } if (request instanceof UpdateRequest) { // All UpdateRequests require a GET to determine the correct mirrored source. pendingRequests.add(request); pendingTargetIndices.add(dualWriteTarget); pendingSlots.add(i); } else if (request instanceof IndexRequest src && needsGetForIndexRequest(src)) { // IndexRequests with custom ID + opType=CREATE or CAS conditions need a GET // to verify the source index state before mirroring. pendingRequests.add(request); pendingTargetIndices.add(dualWriteTarget); pendingSlots.add(i); } else if (request instanceof DeleteRequest src) { // DeleteRequests with version/CAS conditions need a GET to verify the source // state matches before mirroring — if the condition isn't met the delete fails. pendingRequests.add(request); pendingTargetIndices.add(dualWriteTarget); pendingSlots.add(i); } else { // IndexRequest (safe to copy) and DeleteRequest — handled synchronously. final DocWriteRequest<?> mirrored = buildDualWriteRequestSync(request, dualWriteTarget); if (mirrored != null) { bulkRequest.requests.add(mirrored); } } } if (pendingRequests.isEmpty()) { listener.onResponse(null); return; } // Fire all GETs concurrently; append resolved mirrors; invoke listener when all complete. final int pendingCount = pendingRequests.size(); final AtomicInteger remaining = new AtomicInteger(pendingCount); final java.util.concurrent.atomic.AtomicReference<Exception> firstFailure = new java.util.concurrent.atomic.AtomicReference<>(); final java.util.concurrent.ConcurrentLinkedQueue<DocWriteRequest<?>> resolvedMirrors = new java.util.concurrent.ConcurrentLinkedQueue<>(); for (int j = 0; j < pendingCount; j++) { final DocWriteRequest<?> pending = pendingRequests.get(j); final String dualWriteTarget = pendingTargetIndices.get(j); final int slot = pendingSlots.get(j); final GetRequest getRequest = new GetRequest(pending.index(), pending.id()); getRequest.routing(pending.routing()); getRequest.realtime(true); client.execute(GetAction.INSTANCE, getRequest, new ActionListener<GetResponse>() { @Override public void onResponse(GetResponse getResponse) { try { if (pending instanceof UpdateRequest src) { handleUpdateRequestWithGet(src, dualWriteTarget, slot, getResponse, bulkRequest, resolvedMirrors); } else if (pending instanceof IndexRequest src) { handleIndexRequestWithGet(src, dualWriteTarget, getResponse, resolvedMirrors); } else if (pending instanceof DeleteRequest src) { handleDeleteRequestWithGet(src, dualWriteTarget, getResponse, resolvedMirrors); } } catch (Exception e) { firstFailure.compareAndSet(null, e); } finally { maybeFinish(); } } @Override public void onFailure(Exception e) { firstFailure.compareAndSet(null, e); maybeFinish(); } private void maybeFinish() { if (remaining.decrementAndGet() == 0) { final Exception failure = firstFailure.get(); if (failure != null) { listener.onFailure(failure); } else { bulkRequest.requests.addAll(resolvedMirrors); listener.onResponse(null); } } } }); } } Auto-generated ID divergence For IndexRequest with an auto-generated ID (no id set), buildDualWriteRequestSync creates a mirror IndexRequest without setting an id. The mirror will then get its own independently-generated ID on the target index, causing source and target documents to have different _ids. This breaks the ability to correlate documents between source and mirror indices and defeats the purpose of a mirrored copy for later reads/deletes. if (original instanceof IndexRequest src) { IndexRequest mirror = buildDualWriteIndexRequest( dualWriteTarget, src.id(), src.routing(), src.source(), src.getContentType(), src.timeout(), src.getRefreshPolicy(), src.waitForActiveShards() ); // Preserve write semantics specific to direct IndexRequests. mirror.opType(src.opType()); mirror.version(src.version()); mirror.versionType(src.versionType()); return mirror; Preflight GET is not authoritative Preflight GETs are used to decide whether to generate a mirror request, but there is no coordination with the actual write on the source shard. Between the GET and the shard-level write, another concurrent operation can change seq_no/version/existence. As a result: (a) a CAS/create that succeeds on the source may not be mirrored (GET saw a mismatch), or (b) a mirror may be written even though the source write ultimately fails. This creates silent source/target divergence that the current tests do not exercise under concurrency. private void handleDeleteRequestWithGet( DeleteRequest src, String dualWriteTarget, GetResponse getResponse, java.util.concurrent.ConcurrentLinkedQueue<DocWriteRequest<?>> resolvedMirrors ) { if (!getResponse.isExists()) { return; } if (src.ifSeqNo() != UNASSIGNED_SEQ_NO) { if (getResponse.getSeqNo() != src.ifSeqNo() || getResponse.getPrimaryTerm() != src.ifPrimaryTerm()) { return; } } // Version check if (src.version() != Versions.MATCH_ANY) { if (src.versionType().isVersionConflictForWrites(getResponse.getVersion(), src.version(), false)) { return; } } resolvedMirrors.add( buildDualWriteDeleteRequest( dualWriteTarget, src.id(), src.routing(), src.index(), src.timeout(), src.getRefreshPolicy(), src.waitForActiveShards() ) ); } private static boolean needsGetForIndexRequest(IndexRequest src) { if (src.id() == null || src.getAutoGeneratedTimestamp() != IndexRequest.UNSET_AUTO_GENERATED_TIMESTAMP) { return false; } return src.opType() == DocWriteRequest.OpType.CREATE || src.ifSeqNo() != UNASSIGNED_SEQ_NO; } private void handleIndexRequestWithGet( IndexRequest src, String dualWriteTarget, GetResponse getResponse, java.util.concurrent.ConcurrentLinkedQueue<DocWriteRequest<?>> resolvedMirrors ) { if (getResponse.isExists()) { if (src.opType() == DocWriteRequest.OpType.CREATE) { return; } if (src.ifSeqNo() != UNASSIGNED_SEQ_NO) { if (getResponse.getSeqNo() != src.ifSeqNo() || getResponse.getPrimaryTerm() != src.ifPrimaryTerm()) { return; } } } IndexRequest mirror = buildDualWriteIndexRequest( dualWriteTarget, src.id(), src.routing(), src.source(), src.getContentType(), src.timeout(), src.getRefreshPolicy(), src.waitForActiveShards() ); mirror.opType(src.opType()); resolvedMirrors.add(mirror); } UpdateRequest scripts/params dropped handleUpdateRequestWithGet calls updateHelper.prepare at coordinator time using only the fetched GetResult. Script-based updates that depend on runtime state, or updates whose behavior differs when re-executed, will have their resolved IndexRequest/DeleteRequest replace the original UpdateRequest via bulkRequest.requests.set(slot, sourceResolved). This bypasses the shard-level update path entirely, changing semantics: retry_on_conflict is lost, and any script re-evaluation on retry no longer occurs. Consider whether script updates and retry_on_conflict should be excluded from dual-write. private void handleUpdateRequestWithGet( UpdateRequest src, String dualWriteTarget, int slot, GetResponse getResponse, BulkRequest bulkRequest, java.util.concurrent.ConcurrentLinkedQueue<DocWriteRequest<?>> resolvedMirrors ) { if (src.ifSeqNo() != UNASSIGNED_SEQ_NO && getResponse.isExists()) { if (getResponse.getSeqNo() != src.ifSeqNo() || getResponse.getPrimaryTerm() != src.ifPrimaryTerm()) { return; } } final GetResult getResult = new GetResult( getResponse.getIndex(), getResponse.getId(), getResponse.getSeqNo(), getResponse.getPrimaryTerm(), getResponse.getVersion(), getResponse.isExists(), getResponse.isExists() ? getResponse.getSourceInternal() : null, Collections.emptyMap(), Collections.emptyMap() ); final org.opensearch.core.index.shard.ShardId dummyShardId = new org.opensearch.core.index.shard.ShardId(src.index(), "_na_", 0); final UpdateHelper.Result result = updateHelper.prepare(dummyShardId, src, getResult, threadPool::absoluteTimeInMillis); switch (result.getResponseResult()) { case CREATED: case UPDATED: { final IndexRequest sourceResolved = result.<IndexRequest>action(); // Replace the UpdateRequest in the source slot — shard skips the GET. bulkRequest.requests.set(slot, sourceResolved); // Mirror to target (CAS fields stripped). resolvedMirrors.add( buildDualWriteIndexRequest( dualWriteTarget, sourceResolved.id() != null ? sourceResolved.id() : src.id(), sourceResolved.routing(), sourceResolved.source(), sourceResolved.getContentType(), src.timeout(), src.getRefreshPolicy(), src.waitForActiveShards() ) ); break; } case DELETED: { final DeleteRequest sourceResolved = result.<DeleteRequest>action(); bulkRequest.requests.set(slot, sourceResolved); final DocWriteRequest<?> mirror = buildDualWriteDeleteRequest( dualWriteTarget, sourceResolved.id() != null ? sourceResolved.id() : src.id(), sourceResolved.routing(), src.index(), src.timeout(), src.getRefreshPolicy(), src.waitForActiveShards() ); if (mirror != null) { resolvedMirrors.add(mirror); } break; } case NOOP: default: // Noop — leave the original UpdateRequest in place. break; } } Recursive iteration risk expandDualWriteRequests iterates up to originalSize and appends mirrors to the same bulkRequest.requests list. If the dual-write target index itself has index.dual_write.index_name set (misconfiguration or chained mirror), the added mirrors are not iterated here, but on the re-entry after pipeline execution isDualWriteExpanded() prevents re-expansion — which is fine. However, the pipeline re-resolution step (resolvePipelinesForActionRequests) runs on the whole expanded list including mirrors; if a mirror's target has a default pipeline that itself references a dual-write index, there is no defense. Consider validating that the target index does not have dual-write enabled at setting-update time. // Pipelines on the original requests have already executed (or there were none). // Expand dual-write requests now — but only once. The flag on BulkRequest prevents // re-expansion when doInternalExecute is called again after mirror pipeline execution. if (!bulkRequest.isDualWriteExpanded()) { bulkRequest.setDualWriteExpanded(true); expandDualWriteRequests(bulkRequest, metadata, ActionListener.wrap(ignored -> { // Re-resolve pipelines for mirrored requests — target index may have its own pipeline. // processBulkIndexIngestRequest will call doInternalExecute again, but // isDualWriteExpanded()==true so expansion is skipped on that re-entry. boolean hasMirrorPipelines = resolvePipelinesForActionRequests(bulkRequest.requests, metadata, minNodeVersion); if (hasMirrorPipelines) { try { if (clusterService.localNode().isIngestNode()) { processBulkIndexIngestRequest(task, bulkRequest, executorName, listener); } else { ingestForwarder.forwardIngestRequest(BulkAction.INSTANCE, bulkRequest, listener); } } catch (Exception e) { listener.onFailure(e); } return; } dispatchBulk(task, bulkRequest, startTime, executorName, listener, metadata, minNodeVersion); }, listener::onFailure)); return; } // Dual-write expansion already done — all pipelines (source + mirror) have run. // Dispatch to shards. dispatchBulk(task, bulkRequest, startTime, executorName, listener, metadata, minNodeVersion);

[github-actions]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
General	Avoid unnecessary GET for plain deletes All DeleteRequests are unconditionally routed through the async GET path, but only deletes with version/CAS conditions require a GET. Non-conditional deletes should be mirrored synchronously via buildDualWriteRequestSync, matching the intent of needsGetForDeleteRequest (which is currently defined but never called). This avoids an unnecessary GET round-trip per plain delete. server/src/main/java/org/opensearch/action/bulk/TransportBulkAction.java [498-504] } else if (request instanceof IndexRequest src && needsGetForIndexRequest(src)) { // IndexRequests with custom ID + opType=CREATE or CAS conditions need a GET // to verify the source index state before mirroring. pendingRequests.add(request); pendingTargetIndices.add(dualWriteTarget); pendingSlots.add(i); - } else if (request instanceof DeleteRequest src) { + } else if (request instanceof DeleteRequest src && needsGetForDeleteRequest(src)) { // DeleteRequests with version/CAS conditions need a GET to verify the source // state matches before mirroring — if the condition isn't met the delete fails. pendingRequests.add(request); pendingTargetIndices.add(dualWriteTarget); pendingSlots.add(i); } else { Suggestion importance[1-10]: 8 __ Why: Correct observation: needsGetForDeleteRequest is defined but never called, and all DeleteRequests are unconditionally routed through the async GET path, causing unnecessary GETs for plain deletes. This is a meaningful performance and correctness improvement.	Medium
	Clarify source-item selection assumption Blindly returning items[0] when the array can now contain multiple items (source + mirror) may return the mirror's response instead of the source's if ordering ever changes, silently corrupting single-write API responses. Explicitly select the response for the source index (or first non-mirror item) rather than relying on positional index 0, and update the assertion message accordingly. server/src/main/java/org/opensearch/action/bulk/TransportSingleItemBulkWriteAction.java [93-94] -assert bulkItemResponses.getItems().length >= 1 : "expected only one item in bulk request"; +assert bulkItemResponses.getItems().length >= 1 : "expected at least one item in bulk request"; +// Source item is always at index 0; mirror items (if any) follow. BulkItemResponse bulkItemResponse = bulkItemResponses.getItems()[0]; Suggestion importance[1-10]: 5 __ Why: Reasonable observation about assumption fragility, but the improved_code only changes the assertion message without addressing the underlying selection concern, limiting its actual impact.	Low
	Defer mutations to avoid index drift expandDualWriteRequests iterates by index and skips null requests, but pendingSlots stores the original index; later bulkRequest.requests.set(slot, sourceResolved) may overwrite the wrong entry if the request list is mutated concurrently or reordered. Additionally, mirror requests appended synchronously to bulkRequest.requests inside the loop cause the loop bound originalSize to be preserved, but if any code path resizes requests before completion, indexes drift. Snapshot indices only from [0, originalSize) and defer all mutations (both sync mirrors and slot replacement) until after iteration completes to keep index semantics consistent. server/src/main/java/org/opensearch/action/bulk/TransportBulkAction.java [487-511] if (request instanceof UpdateRequest) { - // All UpdateRequests require a GET to determine the correct mirrored source. pendingRequests.add(request); pendingTargetIndices.add(dualWriteTarget); pendingSlots.add(i); Suggestion importance[1-10]: 3 __ Why: The concern about index drift is somewhat speculative; originalSize is captured before iteration and sync mirrors are appended after, not inserted. The suggestion doesn't clearly demonstrate a concrete bug and the improved_code is essentially the same as existing.	Low
Possible issue	Prevent NPE on missing source index metadata metadata.index(sourceIndexName) can return null (e.g. if the source index was deleted between lookups) which will NPE. Guard against a null IndexMetadata and fall back to a safe default so a race with index deletion does not fail the entire bulk. server/src/main/java/org/opensearch/action/bulk/TransportBulkAction.java [786-788] -final boolean softDeletesEnabled = IndexSettings.INDEX_DUAL_WRITE_SOFT_DELETES_ENABLED_SETTING.get( - clusterService.state().metadata().index(sourceIndexName).getSettings() - ); +final IndexMetadata sourceIndexMetadata = clusterService.state().metadata().index(sourceIndexName); +final boolean softDeletesEnabled = sourceIndexMetadata != null + && IndexSettings.INDEX_DUAL_WRITE_SOFT_DELETES_ENABLED_SETTING.get(sourceIndexMetadata.getSettings()); Suggestion importance[1-10]: 6 __ Why: Valid defensive check — metadata.index(sourceIndexName) could return null in a race with index deletion, causing an NPE. Guarding it improves robustness, though the race is unlikely in practice.	Low

[github-actions]

❌ Gradle check result for 34cf093: 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?