addressing comments

bharath-techie · bharath-techie · commit 96c8b3e45212 · 2022-08-18T17:42:04.000+05:30
Signed-off-by: Bharathwaj G &lt;bharath78910@gmail.com&gt;
diff --git a/distribution/build.gradle b/distribution/build.gradle
@@ -621,7 +621,13 @@ subprojects {
   }
 }
 
-['archives:darwin-tar'
+['archives:darwin-tar',
+ 'archives:integ-test-zip',
+ 'archives:linux-arm64-tar',
+ 'archives:linux-tar',
+ 'archives:windows-zip',
+ 'packages:arm64-rpm', 'packages:arm64-deb',
+ 'packages:rpm', 'packages:deb'
 ].forEach { subName ->
   Project subproject = project("${project.path}:${subName}")
   Configuration configuration = configurations.create(subproject.name)
diff --git a/server/src/main/java/org/opensearch/action/ActionModule.java b/server/src/main/java/org/opensearch/action/ActionModule.java
@@ -241,7 +241,6 @@
 import org.opensearch.action.search.GetAllPitsAction;
 import org.opensearch.action.search.MultiSearchAction;
 import org.opensearch.action.search.NodesGetAllPitsAction;
-import org.opensearch.action.search.PitGetAllPitsAction;
 import org.opensearch.action.search.SearchAction;
 import org.opensearch.action.search.SearchScrollAction;
 import org.opensearch.action.search.TransportClearScrollAction;
@@ -250,7 +249,6 @@
 import org.opensearch.action.search.TransportGetAllPitsAction;
 import org.opensearch.action.search.TransportMultiSearchAction;
 import org.opensearch.action.search.TransportNodesGetAllPitsAction;
-import org.opensearch.action.search.TransportPitGetAllPitsAction;
 import org.opensearch.action.search.TransportSearchAction;
 import org.opensearch.action.search.TransportSearchScrollAction;
 import org.opensearch.action.support.ActionFilters;
@@ -683,7 +681,6 @@ public <Request extends ActionRequest, Response extends ActionResponse> void reg
         actions.register(DeletePitAction.INSTANCE, TransportDeletePitAction.class);
         actions.register(PitSegmentsAction.INSTANCE, TransportPitSegmentsAction.class);
         actions.register(NodesGetAllPitsAction.INSTANCE, TransportNodesGetAllPitsAction.class);
-        actions.register(PitGetAllPitsAction.INSTANCE, TransportPitGetAllPitsAction.class);
 
         // Remote Store
         actions.register(RestoreRemoteStoreAction.INSTANCE, TransportRestoreRemoteStoreAction.class);
diff --git a/server/src/main/java/org/opensearch/action/search/DeletePitAction.java b/server/src/main/java/org/opensearch/action/search/DeletePitAction.java
@@ -16,7 +16,7 @@
 public class DeletePitAction extends ActionType<DeletePitResponse> {
 
     public static final DeletePitAction INSTANCE = new DeletePitAction();
-    public static final String NAME = "cluster:admin/pit/delete";
+    public static final String NAME = "cluster:admin/point_in_time/delete";
 
     private DeletePitAction() {
         super(NAME, DeletePitResponse::new);
diff --git a/server/src/main/java/org/opensearch/action/search/GetAllPitsAction.java b/server/src/main/java/org/opensearch/action/search/GetAllPitsAction.java
@@ -15,7 +15,7 @@
  */
 public class GetAllPitsAction extends ActionType<GetAllPitNodesResponse> {
     public static final GetAllPitsAction INSTANCE = new GetAllPitsAction();
-    public static final String NAME = "cluster:admin/pit/read";
+    public static final String NAME = "cluster:admin/point_in_time/read";
 
     private GetAllPitsAction() {
         super(NAME, GetAllPitNodesResponse::new);
diff --git a/server/src/main/java/org/opensearch/action/search/NodesGetAllPitsAction.java b/server/src/main/java/org/opensearch/action/search/NodesGetAllPitsAction.java
@@ -14,7 +14,7 @@
  */
 public class NodesGetAllPitsAction extends ActionType<GetAllPitNodesResponse> {
     public static final NodesGetAllPitsAction INSTANCE = new NodesGetAllPitsAction();
-    public static final String NAME = "cluster:admin/pit/readall";
+    public static final String NAME = "cluster:admin/point_in_time/read_from_nodes";
 
     private NodesGetAllPitsAction() {
         super(NAME, GetAllPitNodesResponse::new);
diff --git a/server/src/main/java/org/opensearch/action/search/PitGetAllPitsAction.java b/server/src/main/java/org/opensearch/action/search/PitGetAllPitsAction.java
diff --git a/server/src/main/java/org/opensearch/action/search/TransportGetAllPitsAction.java b/server/src/main/java/org/opensearch/action/search/TransportGetAllPitsAction.java
@@ -8,79 +8,28 @@
 
 package org.opensearch.action.search;
 
-import org.opensearch.action.FailedNodeException;
+import org.opensearch.action.ActionListener;
 import org.opensearch.action.support.ActionFilters;
-import org.opensearch.action.support.nodes.TransportNodesAction;
-import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.action.support.HandledTransportAction;
 import org.opensearch.common.inject.Inject;
-import org.opensearch.common.io.stream.StreamInput;
-import org.opensearch.search.SearchService;
-import org.opensearch.threadpool.ThreadPool;
+import org.opensearch.tasks.Task;
 import org.opensearch.transport.TransportService;
 
-import java.io.IOException;
-import java.util.List;
-
 /**
- * Transport action to get all active PIT contexts across all nodes
+ * Transport action to get all active PIT contexts in the cluster
  */
-public class TransportGetAllPitsAction extends TransportNodesAction<
-    GetAllPitNodesRequest,
-    GetAllPitNodesResponse,
-    GetAllPitNodeRequest,
-    GetAllPitNodeResponse> {
-    private final SearchService searchService;
-
-    @Inject
-    public TransportGetAllPitsAction(
-        ThreadPool threadPool,
-        ClusterService clusterService,
-        TransportService transportService,
-        ActionFilters actionFilters,
-        SearchService searchService
-    ) {
-        super(
-            GetAllPitsAction.NAME,
-            threadPool,
-            clusterService,
-            transportService,
-            actionFilters,
-            GetAllPitNodesRequest::new,
-            GetAllPitNodeRequest::new,
-            ThreadPool.Names.SAME,
-            GetAllPitNodeResponse.class
-        );
-        this.searchService = searchService;
-    }
+public class TransportGetAllPitsAction extends HandledTransportAction<GetAllPitNodesRequest, GetAllPitNodesResponse> {
 
-    @Override
-    protected GetAllPitNodesResponse newResponse(
-        GetAllPitNodesRequest request,
-        List<GetAllPitNodeResponse> getAllPitNodeRespons,
-        List<FailedNodeException> failures
-    ) {
-        return new GetAllPitNodesResponse(clusterService.getClusterName(), getAllPitNodeRespons, failures);
-    }
+    private final PitService pitService;
 
-    @Override
-    protected GetAllPitNodeRequest newNodeRequest(GetAllPitNodesRequest request) {
-        return new GetAllPitNodeRequest();
-    }
-
-    @Override
-    protected GetAllPitNodeResponse newNodeResponse(StreamInput in) throws IOException {
-        return new GetAllPitNodeResponse(in);
+    @Inject
+    public TransportGetAllPitsAction(ActionFilters actionFilters, TransportService transportService, PitService pitService) {
+        super(GetAllPitsAction.NAME, transportService, actionFilters, in -> new GetAllPitNodesRequest(in));
+        this.pitService = pitService;
     }
 
-    /**
-     * This retrieves all active PITs in the node
-     */
     @Override
-    protected GetAllPitNodeResponse nodeOperation(GetAllPitNodeRequest request) {
-        GetAllPitNodeResponse nodeResponse = new GetAllPitNodeResponse(
-            transportService.getLocalNode(),
-            searchService.getAllPITReaderContexts()
-        );
-        return nodeResponse;
+    protected void doExecute(Task task, GetAllPitNodesRequest request, ActionListener<GetAllPitNodesResponse> listener) {
+        pitService.getAllPits(listener);
     }
 }
diff --git a/server/src/main/java/org/opensearch/action/search/TransportPitGetAllPitsAction.java b/server/src/main/java/org/opensearch/action/search/TransportPitGetAllPitsAction.java
diff --git a/server/src/main/java/org/opensearch/client/support/AbstractClient.java b/server/src/main/java/org/opensearch/client/support/AbstractClient.java
@@ -337,11 +337,11 @@
 import org.opensearch.action.search.DeletePitResponse;
 import org.opensearch.action.search.GetAllPitNodesRequest;
 import org.opensearch.action.search.GetAllPitNodesResponse;
+import org.opensearch.action.search.GetAllPitsAction;
 import org.opensearch.action.search.MultiSearchAction;
 import org.opensearch.action.search.MultiSearchRequest;
 import org.opensearch.action.search.MultiSearchRequestBuilder;
 import org.opensearch.action.search.MultiSearchResponse;
-import org.opensearch.action.search.PitGetAllPitsAction;
 import org.opensearch.action.search.SearchAction;
 import org.opensearch.action.search.SearchRequest;
 import org.opensearch.action.search.SearchRequestBuilder;
@@ -1274,12 +1274,12 @@ public ActionFuture<AcknowledgedResponse> deleteDanglingIndex(DeleteDanglingInde
 
         @Override
         public void getAllPits(GetAllPitNodesRequest request, ActionListener<GetAllPitNodesResponse> listener) {
-            execute(PitGetAllPitsAction.INSTANCE, request, listener);
+            execute(GetAllPitsAction.INSTANCE, request, listener);
         }
 
         @Override
         public ActionFuture<GetAllPitNodesResponse> getAllPits(GetAllPitNodesRequest request) {
-            return execute(PitGetAllPitsAction.INSTANCE, request);
+            return execute(GetAllPitsAction.INSTANCE, request);
         }
 
         @Override
diff --git a/server/src/main/java/org/opensearch/search/SearchService.java b/server/src/main/java/org/opensearch/search/SearchService.java
@@ -881,6 +881,7 @@ public void createPitReaderContext(ShardId shardId, TimeValue keepAlive, ActionL
         shard.awaitShardSearchActive(ignored -> {
             Engine.SearcherSupplier searcherSupplier = null;
             ReaderContext readerContext = null;
+            Releasable decreasePitContexts = openPitContexts::decrementAndGet;
             try {
                 if (openPitContexts.incrementAndGet() > maxOpenPitContext) {
                     throw new OpenSearchRejectedExecutionException(
@@ -902,15 +903,16 @@ public void createPitReaderContext(ShardId shardId, TimeValue keepAlive, ActionL
                 searchOperationListener.onNewPitContext(finalReaderContext);
 
                 readerContext.addOnClose(() -> {
-                    openPitContexts.decrementAndGet();
                     searchOperationListener.onFreeReaderContext(finalReaderContext);
                     searchOperationListener.onFreePitContext(finalReaderContext);
                 });
+                readerContext.addOnClose(decreasePitContexts);
                 // add the newly created pit reader context to active readers
                 putReaderContext(readerContext);
                 readerContext = null;
                 listener.onResponse(finalReaderContext.id());
             } catch (Exception exc) {
+                Releasables.closeWhileHandlingException(decreasePitContexts);
                 Releasables.closeWhileHandlingException(searcherSupplier, readerContext);
                 listener.onFailure(exc);
             }
diff --git a/server/src/test/java/org/opensearch/search/CreatePitSingleNodeTests.java b/server/src/test/java/org/opensearch/search/CreatePitSingleNodeTests.java
@@ -14,6 +14,10 @@
 import org.opensearch.action.search.CreatePitController;
 import org.opensearch.action.search.CreatePitRequest;
 import org.opensearch.action.search.CreatePitResponse;
+import org.opensearch.action.search.DeletePitAction;
+import org.opensearch.action.search.DeletePitInfo;
+import org.opensearch.action.search.DeletePitRequest;
+import org.opensearch.action.search.DeletePitResponse;
 import org.opensearch.action.search.PitTestsUtil;
 import org.opensearch.action.search.SearchPhaseExecutionException;
 import org.opensearch.action.search.SearchResponse;
@@ -33,6 +37,8 @@
 import java.util.concurrent.ExecutionException;
 
 import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.Matchers.blankOrNullString;
+import static org.hamcrest.Matchers.not;
 import static org.opensearch.action.search.PitTestsUtil.assertSegments;
 import static org.opensearch.action.support.WriteRequest.RefreshPolicy.IMMEDIATE;
 import static org.opensearch.common.xcontent.XContentFactory.jsonBuilder;
@@ -277,6 +283,24 @@ public void testMaxOpenPitContexts() throws Exception {
                 )
         );
         final int maxPitContexts = SearchService.MAX_OPEN_PIT_CONTEXT.get(Settings.EMPTY);
+        DeletePitRequest deletePITRequest = new DeletePitRequest("_all");
+
+        /**
+         * When we invoke delete again, returns success after clearing the remaining readers. Asserting reader context
+         * not found exceptions don't result in failures ( as deletion in one node is successful )
+         */
+        ActionFuture<DeletePitResponse> execute1 = client().execute(DeletePitAction.INSTANCE, deletePITRequest);
+        DeletePitResponse deletePITResponse = execute1.get();
+        for (DeletePitInfo deletePitInfo : deletePITResponse.getDeletePitResults()) {
+            assertThat(deletePitInfo.getPitId(), not(blankOrNullString()));
+            assertTrue(deletePitInfo.isSuccessful());
+        }
+        assertEquals(0,service.getActiveContexts());
+        execute = client().execute(CreatePitAction.INSTANCE, request);
+        CreatePitResponse pitResponse = execute.get();
+        assertEquals(1,service.getActiveContexts());
+
+        validatePitStats("index", 0, maxPitContexts, 0);
         validatePitStats("index", maxPitContexts, 0, 0);
         service.doClose();
         validatePitStats("index", 0, maxPitContexts, 0);

Original file line number	Diff line number	Diff line change
`@@ -621,7 +621,13 @@ subprojects {`
`621`	`621`	`}`
`622`	`622`	`}`
`623`	`623`
`624`		`-['archives:darwin-tar'`
	`624`	`+['archives:darwin-tar',`
	`625`	`+ 'archives:integ-test-zip',`
	`626`	`+ 'archives:linux-arm64-tar',`
	`627`	`+ 'archives:linux-tar',`
	`628`	`+ 'archives:windows-zip',`
	`629`	`+ 'packages:arm64-rpm', 'packages:arm64-deb',`
	`630`	`+ 'packages:rpm', 'packages:deb'`
`625`	`631`	`].forEach { subName ->`
`626`	`632`	`Project subproject = project("${project.path}:${subName}")`
`627`	`633`	`Configuration configuration = configurations.create(subproject.name)`