PR Feedback:

- Add random refresh policy to reloation ITs. - add back finishing ReplicationTask. Signed-off-by: Marc Handalian <handalm@amazon.com>
opensearch-project · Feb 20, 2023 · e360577 · e360577
1 parent 33edf00
commit e360577
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 5 deletions.
diff --git a/...alClusterTest/java/org/opensearch/indices/replication/SegmentReplicationRelocationIT.java b/...alClusterTest/java/org/opensearch/indices/replication/SegmentReplicationRelocationIT.java
@@ -55,12 +55,13 @@ public void testPrimaryRelocation() throws Exception {
         final String replica = internalCluster().startNode();
         ensureGreen(INDEX_NAME);
         final int initialDocCount = scaledRandomIntBetween(100, 1000);
+        final WriteRequest.RefreshPolicy refreshPolicy = randomFrom(WriteRequest.RefreshPolicy.values());
         final List<ActionFuture<IndexResponse>> pendingIndexResponses = new ArrayList<>();
         for (int i = 0; i < initialDocCount; i++) {
             pendingIndexResponses.add(
                 client().prepareIndex(INDEX_NAME)
                     .setId(Integer.toString(i))
-                    .setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL)
+                    .setRefreshPolicy(refreshPolicy)
                     .setSource("field", "value" + i)
                     .execute()
             );
@@ -105,7 +106,7 @@ public void testPrimaryRelocation() throws Exception {
             pendingIndexResponses.add(
                 client().prepareIndex(INDEX_NAME)
                     .setId(Integer.toString(i))
-                    .setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL)
+                    .setRefreshPolicy(refreshPolicy)
                     .setSource("field", "value" + i)
                     .execute()
             );
@@ -131,12 +132,13 @@ public void testPrimaryRelocationWithSegRepFailure() throws Exception {
         final String replica = internalCluster().startNode();
         ensureGreen(INDEX_NAME);
         final int initialDocCount = scaledRandomIntBetween(100, 1000);
+        final WriteRequest.RefreshPolicy refreshPolicy = randomFrom(WriteRequest.RefreshPolicy.values());
         final List<ActionFuture<IndexResponse>> pendingIndexResponses = new ArrayList<>();
         for (int i = 0; i < initialDocCount; i++) {
             pendingIndexResponses.add(
                 client().prepareIndex(INDEX_NAME)
                     .setId(Integer.toString(i))
-                    .setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL)
+                    .setRefreshPolicy(refreshPolicy)
                     .setSource("field", "value" + i)
                     .execute()
             );
@@ -189,7 +191,7 @@ public void testPrimaryRelocationWithSegRepFailure() throws Exception {
             pendingIndexResponses.add(
                 client().prepareIndex(INDEX_NAME)
                     .setId(Integer.toString(i))
-                    .setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL)
+                    .setRefreshPolicy(refreshPolicy)
                     .setSource("field", "value" + i)
                     .execute()
             );
@@ -297,14 +299,15 @@ public void testRelocateWithQueuedOperationsDuringHandoff() throws Exception {
         }
         logger.info("--> flush to have segments on disk");
         client().admin().indices().prepareFlush().execute().actionGet();
+        final WriteRequest.RefreshPolicy refreshPolicy = randomFrom(WriteRequest.RefreshPolicy.values());
 
         logger.info("--> index more docs so there are ops in the transaction log");
         final List<ActionFuture<IndexResponse>> pendingIndexResponses = new ArrayList<>();
         for (int i = 10; i < 20; i++) {
             pendingIndexResponses.add(
                 client().prepareIndex(INDEX_NAME)
                     .setId(Integer.toString(i))
-                    .setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL)
+                    .setRefreshPolicy(refreshPolicy)
                     .setSource("field", "value" + i)
                     .execute()
             );

diff --git a/.../src/main/java/org/opensearch/indices/replication/checkpoint/PublishCheckpointAction.java b/.../src/main/java/org/opensearch/indices/replication/checkpoint/PublishCheckpointAction.java
@@ -146,6 +146,7 @@ public void onResponse(ReplicationOperation.ReplicaResponse replicaResponse) {
                                 timer.time()
                             )
                         );
+                        task.setPhase("finished");
                         taskManager.unregister(task);
                     }