Skip to content

Commit 4b114c0

Browse files
atrisatris
authored
When update operations fail during preparation (e.g., version conflicts), (opensearch-project#18917)
TransportShardBulkAction still triggers refresh even though no actual writes occurred. This fix checks if locationToSync is null (indicating no writes) and prevents refresh in such cases. Fixes opensearch-project#15261 Signed-off-by: Atri Sharma <atri.jiit@gmail.com>
1 parent 292a2c5 commit 4b114c0

3 files changed

Lines changed: 292 additions & 2 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
6868
- Remove `experimental` designation from transport-grpc settings ([#18915](https://github.com/opensearch-project/OpenSearch/pull/18915))
6969
- Rename package org.opensearch.plugin,transport.grpc to org.opensearch.transport.grpc ([#18923](https://github.com/opensearch-project/OpenSearch/pull/18923))
7070

71+
### Fixed
72+
- Fix unnecessary refreshes on update preparation failures ([#15261](https://github.com/opensearch-project/OpenSearch/issues/15261))
73+
7174
### Dependencies
7275
- Bump `stefanzweifel/git-auto-commit-action` from 5 to 6 ([#18524](https://github.com/opensearch-project/OpenSearch/pull/18524))
7376
- Bump Apache Lucene to 10.2.2 ([#18573](https://github.com/opensearch-project/OpenSearch/pull/18573))

‎server/src/main/java/org/opensearch/action/bulk/TransportShardBulkAction.java‎

Lines changed: 23 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,7 @@
4747
import org.opensearch.action.index.IndexResponse;
4848
import org.opensearch.action.support.ActionFilters;
4949
import org.opensearch.action.support.ChannelActionListener;
50+
import org.opensearch.action.support.WriteRequest;
5051
import org.opensearch.action.support.replication.ReplicationMode;
5152
import org.opensearch.action.support.replication.ReplicationOperation;
5253
import org.opensearch.action.support.replication.ReplicationTask;
@@ -520,12 +521,32 @@ public boolean isForceExecution() {
520521
}
521522

522523
private void finishRequest() {
524+
// If no actual writes occurred (locationToSync is null), we should not trigger refresh
525+
// even if the request has RefreshPolicy.IMMEDIATE
526+
final Translog.Location locationToSync = context.getLocationToSync();
527+
final BulkShardRequest bulkShardRequest = context.getBulkShardRequest();
528+
529+
// Create a modified request with NONE refresh policy if no writes occurred
530+
final BulkShardRequest requestForResult;
531+
if (locationToSync == null && bulkShardRequest.getRefreshPolicy() != WriteRequest.RefreshPolicy.NONE) {
532+
// No actual writes occurred, so we should not refresh
533+
requestForResult = new BulkShardRequest(
534+
bulkShardRequest.shardId(),
535+
WriteRequest.RefreshPolicy.NONE,
536+
bulkShardRequest.items()
537+
);
538+
requestForResult.index(bulkShardRequest.index());
539+
requestForResult.setParentTask(bulkShardRequest.getParentTask());
540+
} else {
541+
requestForResult = bulkShardRequest;
542+
}
543+
523544
ActionListener.completeWith(
524545
listener,
525546
() -> new WritePrimaryResult<>(
526-
context.getBulkShardRequest(),
547+
requestForResult,
527548
context.buildShardResponse(),
528-
context.getLocationToSync(),
549+
locationToSync,
529550
null,
530551
context.getPrimary(),
531552
logger

‎server/src/test/java/org/opensearch/action/bulk/TransportShardBulkActionTests.java‎

Lines changed: 266 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -104,6 +104,7 @@
104104
import java.util.concurrent.CountDownLatch;
105105
import java.util.concurrent.CyclicBarrier;
106106
import java.util.concurrent.TimeUnit;
107+
import java.util.concurrent.atomic.AtomicBoolean;
107108
import java.util.concurrent.atomic.AtomicInteger;
108109
import java.util.function.LongSupplier;
109110

@@ -119,6 +120,7 @@
119120
import static org.mockito.Mockito.any;
120121
import static org.mockito.Mockito.anyBoolean;
121122
import static org.mockito.Mockito.anyLong;
123+
import static org.mockito.Mockito.doAnswer;
122124
import static org.mockito.Mockito.eq;
123125
import static org.mockito.Mockito.mock;
124126
import static org.mockito.Mockito.spy;
@@ -822,6 +824,270 @@ public void testFailureDuringUpdateProcessing() throws Exception {
822824
assertThat(failure.getStatus(), equalTo(RestStatus.INTERNAL_SERVER_ERROR));
823825
}
824826

827+
public void testFailedUpdatePreparationDoesNotTriggerRefresh() throws Exception {
828+
IndexSettings indexSettings = new IndexSettings(indexMetadata(), Settings.EMPTY);
829+
830+
// Create an update request that will fail during preparation
831+
DocWriteRequest<UpdateRequest> writeRequest = new UpdateRequest("index", "id").doc(Requests.INDEX_CONTENT_TYPE, "field", "value");
832+
BulkItemRequest primaryRequest = new BulkItemRequest(0, writeRequest);
833+
834+
IndexShard shard = mock(IndexShard.class);
835+
when(shard.indexSettings()).thenReturn(indexSettings);
836+
when(shard.shardId()).thenReturn(shardId);
837+
838+
// Mock the UpdateHelper to throw a version conflict exception during preparation
839+
UpdateHelper updateHelper = mock(UpdateHelper.class);
840+
final VersionConflictEngineException versionConflict = new VersionConflictEngineException(
841+
shardId,
842+
"id",
843+
"version conflict during update preparation"
844+
);
845+
when(updateHelper.prepare(any(), eq(shard), any())).thenThrow(versionConflict);
846+
847+
// Create bulk request with IMMEDIATE refresh policy
848+
BulkItemRequest[] items = new BulkItemRequest[] { primaryRequest };
849+
BulkShardRequest bulkShardRequest = new BulkShardRequest(shardId, RefreshPolicy.IMMEDIATE, items);
850+
851+
randomlySetIgnoredPrimaryResponse(primaryRequest);
852+
853+
// Execute the bulk operation through performOnPrimary
854+
CountDownLatch latch = new CountDownLatch(1);
855+
AtomicBoolean refreshCalled = new AtomicBoolean(false);
856+
857+
// Mock refresh to track if it's called
858+
doAnswer(invocation -> {
859+
refreshCalled.set(true);
860+
return null;
861+
}).when(shard).refresh(any());
862+
863+
TransportShardBulkAction.performOnPrimary(
864+
bulkShardRequest,
865+
shard,
866+
updateHelper,
867+
threadPool::absoluteTimeInMillis,
868+
new NoopMappingUpdatePerformer(),
869+
listener -> listener.onResponse(null),
870+
new LatchedActionListener<>(ActionTestUtils.assertNoFailureListener(result -> {
871+
WritePrimaryResult<BulkShardRequest, BulkShardResponse> primaryResult = (WritePrimaryResult<
872+
BulkShardRequest,
873+
BulkShardResponse>) result;
874+
875+
// Verify no location to sync (no writes occurred)
876+
assertNull(primaryResult.location);
877+
878+
// Run post replication actions
879+
primaryResult.runPostReplicationActions(ActionListener.wrap(v -> {
880+
// Success - refresh should not have been called
881+
}, e -> { fail("Post replication actions should not fail: " + e.getMessage()); }));
882+
883+
// Verify refresh was NOT called even though refresh policy was IMMEDIATE
884+
assertFalse(refreshCalled.get());
885+
}), latch),
886+
threadPool,
887+
Names.WRITE
888+
);
889+
890+
assertTrue(latch.await(5, TimeUnit.SECONDS));
891+
}
892+
893+
public void testBulkRequestWithMixedSuccessAndFailureRefresh() throws Exception {
894+
IndexSettings indexSettings = new IndexSettings(indexMetadata(), Settings.EMPTY);
895+
896+
// Create a mix of successful and failed operations
897+
BulkItemRequest[] items = new BulkItemRequest[3];
898+
899+
// Item 0: Successful index operation
900+
items[0] = new BulkItemRequest(0, new IndexRequest("index").id("success1").source(Requests.INDEX_CONTENT_TYPE, "field", "value"));
901+
902+
// Item 1: Failed update operation
903+
items[1] = new BulkItemRequest(1, new UpdateRequest("index", "fail1").doc(Requests.INDEX_CONTENT_TYPE, "field", "value"));
904+
905+
// Item 2: Successful index operation
906+
items[2] = new BulkItemRequest(2, new IndexRequest("index").id("success2").source(Requests.INDEX_CONTENT_TYPE, "field", "value"));
907+
908+
BulkShardRequest bulkShardRequest = new BulkShardRequest(shardId, RefreshPolicy.IMMEDIATE, items);
909+
910+
IndexShard shard = mock(IndexShard.class);
911+
when(shard.indexSettings()).thenReturn(indexSettings);
912+
when(shard.shardId()).thenReturn(shardId);
913+
914+
// Mock successful index operations
915+
Translog.Location resultLocation1 = new Translog.Location(42, 42, 42);
916+
Translog.Location resultLocation2 = new Translog.Location(43, 43, 43);
917+
Engine.IndexResult successResult1 = new FakeIndexResult(1, 1, 10, true, resultLocation1);
918+
Engine.IndexResult successResult2 = new FakeIndexResult(1, 1, 12, true, resultLocation2);
919+
920+
when(shard.applyIndexOperationOnPrimary(anyLong(), any(), any(), anyLong(), anyLong(), anyLong(), anyBoolean())).thenReturn(
921+
successResult1
922+
).thenReturn(successResult2);
923+
924+
// Mock failed update operation
925+
UpdateHelper updateHelper = mock(UpdateHelper.class);
926+
when(updateHelper.prepare(any(), eq(shard), any())).thenThrow(
927+
new VersionConflictEngineException(shardId, "fail1", "version conflict")
928+
);
929+
930+
// Track refresh calls
931+
AtomicBoolean refreshCalled = new AtomicBoolean(false);
932+
doAnswer(invocation -> {
933+
refreshCalled.set(true);
934+
return null;
935+
}).when(shard).refresh(any());
936+
937+
// Execute bulk operation
938+
CountDownLatch latch = new CountDownLatch(1);
939+
TransportShardBulkAction.performOnPrimary(
940+
bulkShardRequest,
941+
shard,
942+
updateHelper,
943+
threadPool::absoluteTimeInMillis,
944+
new NoopMappingUpdatePerformer(),
945+
listener -> listener.onResponse(null),
946+
new LatchedActionListener<>(ActionTestUtils.assertNoFailureListener(result -> {
947+
WritePrimaryResult<BulkShardRequest, BulkShardResponse> primaryResult = (WritePrimaryResult<
948+
BulkShardRequest,
949+
BulkShardResponse>) result;
950+
951+
// Should have a location since some operations succeeded
952+
assertNotNull(primaryResult.location);
953+
954+
// Run post replication actions
955+
primaryResult.runPostReplicationActions(ActionListener.wrap(v -> {
956+
// Success
957+
}, e -> { fail("Post replication actions should not fail: " + e.getMessage()); }));
958+
959+
// Verify refresh WAS called because there were successful writes
960+
assertTrue(refreshCalled.get());
961+
}), latch),
962+
threadPool,
963+
Names.WRITE
964+
);
965+
966+
assertTrue(latch.await(5, TimeUnit.SECONDS));
967+
}
968+
969+
public void testBulkRequestWithAllFailedUpdatesNoRefresh() throws Exception {
970+
IndexSettings indexSettings = new IndexSettings(indexMetadata(), Settings.EMPTY);
971+
972+
// Create multiple failed update operations
973+
BulkItemRequest[] items = new BulkItemRequest[3];
974+
for (int i = 0; i < 3; i++) {
975+
items[i] = new BulkItemRequest(i, new UpdateRequest("index", "id" + i).doc(Requests.INDEX_CONTENT_TYPE, "field", "value"));
976+
}
977+
978+
BulkShardRequest bulkShardRequest = new BulkShardRequest(shardId, RefreshPolicy.IMMEDIATE, items);
979+
980+
IndexShard shard = mock(IndexShard.class);
981+
when(shard.indexSettings()).thenReturn(indexSettings);
982+
when(shard.shardId()).thenReturn(shardId);
983+
984+
// Mock all updates to fail
985+
UpdateHelper updateHelper = mock(UpdateHelper.class);
986+
when(updateHelper.prepare(any(), eq(shard), any())).thenThrow(
987+
new VersionConflictEngineException(shardId, "id", "version conflict")
988+
);
989+
990+
// Track refresh calls
991+
AtomicBoolean refreshCalled = new AtomicBoolean(false);
992+
doAnswer(invocation -> {
993+
refreshCalled.set(true);
994+
return null;
995+
}).when(shard).refresh(any());
996+
997+
// Execute bulk operation
998+
CountDownLatch latch = new CountDownLatch(1);
999+
TransportShardBulkAction.performOnPrimary(
1000+
bulkShardRequest,
1001+
shard,
1002+
updateHelper,
1003+
threadPool::absoluteTimeInMillis,
1004+
new NoopMappingUpdatePerformer(),
1005+
listener -> listener.onResponse(null),
1006+
new LatchedActionListener<>(ActionTestUtils.assertNoFailureListener(result -> {
1007+
WritePrimaryResult<BulkShardRequest, BulkShardResponse> primaryResult = (WritePrimaryResult<
1008+
BulkShardRequest,
1009+
BulkShardResponse>) result;
1010+
1011+
// No location since all operations failed
1012+
assertNull(primaryResult.location);
1013+
1014+
// Run post replication actions
1015+
primaryResult.runPostReplicationActions(ActionListener.wrap(v -> {
1016+
// Success
1017+
}, e -> { fail("Post replication actions should not fail: " + e.getMessage()); }));
1018+
1019+
// Verify refresh was NOT called
1020+
assertFalse(refreshCalled.get());
1021+
}), latch),
1022+
threadPool,
1023+
Names.WRITE
1024+
);
1025+
1026+
assertTrue(latch.await(5, TimeUnit.SECONDS));
1027+
}
1028+
1029+
public void testSuccessfulBulkOperationStillTriggersRefresh() throws Exception {
1030+
IndexSettings indexSettings = new IndexSettings(indexMetadata(), Settings.EMPTY);
1031+
1032+
// Create successful operations
1033+
BulkItemRequest[] items = new BulkItemRequest[2];
1034+
items[0] = new BulkItemRequest(0, new IndexRequest("index").id("id1").source(Requests.INDEX_CONTENT_TYPE, "field", "value1"));
1035+
items[1] = new BulkItemRequest(1, new IndexRequest("index").id("id2").source(Requests.INDEX_CONTENT_TYPE, "field", "value2"));
1036+
1037+
BulkShardRequest bulkShardRequest = new BulkShardRequest(shardId, RefreshPolicy.IMMEDIATE, items);
1038+
1039+
IndexShard shard = mock(IndexShard.class);
1040+
when(shard.indexSettings()).thenReturn(indexSettings);
1041+
when(shard.shardId()).thenReturn(shardId);
1042+
1043+
// Mock successful operations
1044+
AtomicInteger locationCounter = new AtomicInteger(42);
1045+
when(shard.applyIndexOperationOnPrimary(anyLong(), any(), any(), anyLong(), anyLong(), anyLong(), anyBoolean())).thenAnswer(
1046+
invocation -> {
1047+
int loc = locationCounter.getAndIncrement();
1048+
return new FakeIndexResult(1, 1, loc, true, new Translog.Location(loc, loc, loc));
1049+
}
1050+
);
1051+
1052+
// Track refresh calls
1053+
AtomicBoolean refreshCalled = new AtomicBoolean(false);
1054+
doAnswer(invocation -> {
1055+
refreshCalled.set(true);
1056+
return null;
1057+
}).when(shard).refresh(any());
1058+
1059+
// Execute bulk operation
1060+
CountDownLatch latch = new CountDownLatch(1);
1061+
TransportShardBulkAction.performOnPrimary(
1062+
bulkShardRequest,
1063+
shard,
1064+
null,
1065+
threadPool::absoluteTimeInMillis,
1066+
new NoopMappingUpdatePerformer(),
1067+
listener -> listener.onResponse(null),
1068+
new LatchedActionListener<>(ActionTestUtils.assertNoFailureListener(result -> {
1069+
WritePrimaryResult<BulkShardRequest, BulkShardResponse> primaryResult = (WritePrimaryResult<
1070+
BulkShardRequest,
1071+
BulkShardResponse>) result;
1072+
1073+
// Should have location from successful operations
1074+
assertNotNull(primaryResult.location);
1075+
1076+
// Run post replication actions
1077+
primaryResult.runPostReplicationActions(ActionListener.wrap(v -> {
1078+
// Success
1079+
}, e -> { fail("Post replication actions should not fail: " + e.getMessage()); }));
1080+
1081+
// Verify refresh WAS called
1082+
assertTrue(refreshCalled.get());
1083+
}), latch),
1084+
threadPool,
1085+
Names.WRITE
1086+
);
1087+
1088+
assertTrue(latch.await(5, TimeUnit.SECONDS));
1089+
}
1090+
8251091
public void testTranslogPositionToSync() throws Exception {
8261092
IndexShard shard = newStartedShard(true);
8271093

0 commit comments

Comments
 (0)