Added multiple items in UT with different retry counts and item level…

… retry assertion Signed-off-by: Raghuvansh Raj <raghraaj@amazon.com>
opensearch-project · Nov 23, 2023 · 6356185 · 6356185
1 parent eb5def5
commit 6356185
Show file tree

Hide file tree

Showing 2 changed files with 34 additions and 18 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -23,7 +23,6 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Allow to pass the list settings through environment variables (like [], ["a", "b", "c"], ...) ([#10625](https://github.com/opensearch-project/OpenSearch/pull/10625))
 - [Remote cluster state] Restore cluster state version during remote state auto restore ([#10853](https://github.com/opensearch-project/OpenSearch/pull/10853))
 - Add back half_float BKD based sort query optimization ([#11024](https://github.com/opensearch-project/OpenSearch/pull/11024))
-- Bugfix for update staying stuck when sent as part of bulk with `retry_on_conflict` specified ([#11152](https://github.com/opensearch-project/OpenSearch/issues/11152))
 
 ### Dependencies
 - Bump `log4j-core` from 2.18.0 to 2.19.0

diff --git a/server/src/test/java/org/opensearch/action/bulk/TransportShardBulkActionTests.java b/server/src/test/java/org/opensearch/action/bulk/TransportShardBulkActionTests.java
@@ -97,12 +97,15 @@
 import org.opensearch.transport.TransportService;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collections;
+import java.util.List;
 import java.util.concurrent.BrokenBarrierException;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.LongSupplier;
 
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.instanceOf;
@@ -950,9 +953,16 @@ public void testRetries() throws Exception {
 
     public void testUpdateWithRetryOnConflict() throws IOException, InterruptedException {
         IndexSettings indexSettings = new IndexSettings(indexMetadata(), Settings.EMPTY);
-        UpdateRequest writeRequest = new UpdateRequest("index", "id").doc(Requests.INDEX_CONTENT_TYPE, "field", "value");
-        writeRequest.retryOnConflict(3);
-        BulkItemRequest primaryRequest = new BulkItemRequest(0, writeRequest);
+
+        int nItems = randomIntBetween(2, 5);
+        List<BulkItemRequest> items = new ArrayList<>(nItems);
+        for (int i = 0; i < nItems; i++) {
+            int retryOnConflictCount = randomIntBetween(0, 3);
+            logger.debug("Setting retryCount for item {}: {}", i, retryOnConflictCount);
+            UpdateRequest updateRequest = new UpdateRequest("index", "id").doc(Requests.INDEX_CONTENT_TYPE, "field", "value")
+                .retryOnConflict(retryOnConflictCount);
+            items.add(new BulkItemRequest(i, updateRequest));
+        }
 
         IndexRequest updateResponse = new IndexRequest("index").id("id").source(Requests.INDEX_CONTENT_TYPE, "field", "value");
 
@@ -977,8 +987,7 @@ public void testUpdateWithRetryOnConflict() throws IOException, InterruptedExcep
             )
         );
 
-        BulkItemRequest[] items = new BulkItemRequest[] { primaryRequest };
-        BulkShardRequest bulkShardRequest = new BulkShardRequest(shardId, RefreshPolicy.NONE, items);
+        BulkShardRequest bulkShardRequest = new BulkShardRequest(shardId, RefreshPolicy.NONE, items.toArray(BulkItemRequest[]::new));
 
         final CountDownLatch latch = new CountDownLatch(1);
         Runnable runnable = () -> TransportShardBulkAction.performOnPrimary(
@@ -988,26 +997,34 @@ public void testUpdateWithRetryOnConflict() throws IOException, InterruptedExcep
             threadPool::absoluteTimeInMillis,
             new NoopMappingUpdatePerformer(),
             listener -> listener.onResponse(null),
-            new LatchedActionListener<>(
-                ActionTestUtils.assertNoFailureListener(
-                    result -> assertEquals(
-                        VersionConflictEngineException.class,
-                        result.replicaRequest().items()[0].getPrimaryResponse().getFailure().getCause().getClass()
-                    )
-                ),
-                latch
-            ),
+            new LatchedActionListener<>(ActionTestUtils.assertNoFailureListener(result -> {
+                assertEquals(nItems, result.replicaRequest().items().length);
+                for (BulkItemRequest item : result.replicaRequest().items()) {
+                    assertEquals(VersionConflictEngineException.class, item.getPrimaryResponse().getFailure().getCause().getClass());
+                }
+            }), latch),
             threadPool,
             Names.WRITE
         );
 
         // execute the runnable on a separate thread so that the infinite loop can be detected
-        Thread thread = new Thread(runnable);
-        thread.start();
+        new Thread(runnable).start();
 
         // timeout the request in 10 seconds if there is an infinite loop
         assertTrue(latch.await(10, TimeUnit.SECONDS));
-        assertEquals(items[0].getPrimaryResponse().getFailure().getCause().getClass(), VersionConflictEngineException.class);
+
+        items.forEach(item -> {
+            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();
+            verify(updateHelper, times(updateRequest.retryOnConflict() + 1)).prepare(
+                eq(updateRequest),
+                any(IndexShard.class),
+                any(LongSupplier.class)
+            );
+        });
     }
 
     public void testForceExecutionOnRejectionAfterMappingUpdate() throws Exception {