addressing comments

Signed-off-by: Bharathwaj G <[email protected]>

client/rest-high-level/src/main/java/org/opensearch/client/RequestConverters.java

@@ -439,7 +439,7 @@ static void addSearchRequestParams(Params params, SearchRequest searchRequest) {
             params.withIndicesOptions(searchRequest.indicesOptions());
         }
         params.withSearchType(searchRequest.searchType().name().toLowerCase(Locale.ROOT));
-        if (searchRequest.pointInTimeBuilder() == null) {
+        if (searchRequest.pointInTimeBuilder() != null) {
             params.putParam("ccs_minimize_roundtrips", "false");
         } else {
             params.putParam("ccs_minimize_roundtrips", Boolean.toString(searchRequest.isCcsMinimizeRoundtrips()));

server/src/main/java/org/opensearch/action/search/CreatePitController.java

@@ -103,7 +103,7 @@ public void executeCreatePit(StepListener<SearchResponse> createPitListener, Act
          * Phase 2 of create PIT where we update pit id in pit contexts
          */
         createPitListener.whenComplete(
-            searchResponse -> { executeUpdatePitId(request, searchResponse, updatePitIdListener); },
+            searchResponse -> { executeUpdatePitId(request, searchRequest, searchResponse, updatePitIdListener); },
             updatePitIdListener::onFailure
         );
     }
@@ -148,6 +148,7 @@ public void executeOnShardTarget(
      */
     void executeUpdatePitId(
         CreatePitRequest request,
+        SearchRequest searchRequest,
         SearchResponse searchResponse,
         ActionListener<CreatePitResponse> updatePitIdListener
     ) {
@@ -161,7 +162,13 @@ void executeUpdatePitId(
          * store the create time ( same create time for all PIT contexts across shards ) to be used
          * for list PIT api
          */
-        final long creationTime = System.currentTimeMillis();
+        final long relativeStartNanos = System.nanoTime();
+        final TransportSearchAction.SearchTimeProvider timeProvider = new TransportSearchAction.SearchTimeProvider(
+            searchRequest.getOrCreateAbsoluteStartMillis(),
+            relativeStartNanos,
+            System::nanoTime
+        );
+        final long creationTime = timeProvider.getAbsoluteStartMillis();
         CreatePitResponse createPITResponse = new CreatePitResponse(
             searchResponse.pointInTimeId(),
             creationTime,
@@ -212,27 +219,17 @@ void executeUpdatePitId(
                 }
             }
         }, updatePitIdListener::onFailure);
-
     }
 
     private StepListener<BiFunction<String, String, DiscoveryNode>> getConnectionLookupListener(SearchContextId contextId) {
         ClusterState state = clusterService.state();
-
         final Set<String> clusters = contextId.shards()
             .values()
             .stream()
             .filter(ctx -> Strings.isEmpty(ctx.getClusterAlias()) == false)
             .map(SearchContextIdForNode::getClusterAlias)
             .collect(Collectors.toSet());
-
-        final StepListener<BiFunction<String, String, DiscoveryNode>> lookupListener = new StepListener<>();
-
-        if (clusters.isEmpty()) {
-            lookupListener.onResponse((cluster, nodeId) -> state.getNodes().get(nodeId));
-        } else {
-            searchTransportService.getRemoteClusterService().collectNodes(clusters, lookupListener);
-        }
-        return lookupListener;
+        return SearchUtils.getConnectionLookupListener(searchTransportService, state, clusters);
     }
 
     private ActionListener<UpdatePitContextResponse> getGroupedListener(

server/src/main/java/org/opensearch/action/search/SearchUtils.java

@@ -0,0 +1,42 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.action.search;
+
+import org.opensearch.action.StepListener;
+import org.opensearch.cluster.ClusterState;
+import org.opensearch.cluster.node.DiscoveryNode;
+
+import java.util.Set;
+import java.util.function.BiFunction;
+
+/**
+ * Helper class for common search functions
+ */
+public class SearchUtils {
+
+    public SearchUtils() {}
+
+    /**
+     * Get connection lookup listener for list of clusters passed
+     */
+    public static StepListener<BiFunction<String, String, DiscoveryNode>> getConnectionLookupListener(
+        SearchTransportService searchTransportService,
+        ClusterState state,
+        Set<String> clusters
+    ) {
+        final StepListener<BiFunction<String, String, DiscoveryNode>> lookupListener = new StepListener<>();
+
+        if (clusters.isEmpty()) {
+            lookupListener.onResponse((cluster, nodeId) -> state.getNodes().get(nodeId));
+        } else {
+            searchTransportService.getRemoteClusterService().collectNodes(clusters, lookupListener);
+        }
+        return lookupListener;
+    }
+}

server/src/main/java/org/opensearch/action/search/TransportCreatePitAction.java

@@ -82,6 +82,9 @@ protected void doExecute(Task task, CreatePitRequest request, ActionListener<Cre
         controller.executeCreatePit(createPitListener, updatePitIdListener);
     }
 
+    /**
+     * Request to create pit reader context with keep alive
+     */
     public static class CreateReaderContextRequest extends TransportRequest {
         private final ShardId shardId;
         private final TimeValue keepAlive;
@@ -113,6 +116,9 @@ public void writeTo(StreamOutput out) throws IOException {
         }
     }
 
+    /**
+     * Create pit reader context response which holds the contextId
+     */
     public static class CreateReaderContextResponse extends SearchPhaseResult {
         public CreateReaderContextResponse(ShardSearchContextId shardSearchContextId) {
             this.contextId = shardSearchContextId;

server/src/main/java/org/opensearch/search/SearchService.java

@@ -869,13 +869,9 @@ public void createPitReaderContext(ShardId shardId, TimeValue keepAlive, ActionL
         final IndexShard shard = indexService.getShard(shardId.id());
         final SearchOperationListener searchOperationListener = shard.getSearchOperationListener();
         shard.awaitShardSearchActive(ignored -> {
-            Releasable decreasePitContexts = null;
             Engine.SearcherSupplier searcherSupplier = null;
             ReaderContext readerContext = null;
-            boolean success = false;
             try {
-                // use this when reader context is freed
-                decreasePitContexts = openPitContexts::decrementAndGet;
                 if (openPitContexts.incrementAndGet() > maxOpenPitContext) {
                     throw new OpenSearchRejectedExecutionException(
                         "Trying to create too many Point In Time contexts. Must be less than or equal to: ["
@@ -894,26 +890,19 @@ public void createPitReaderContext(ShardId shardId, TimeValue keepAlive, ActionL
 
                 searchOperationListener.onNewReaderContext(readerContext);
                 searchOperationListener.onNewPitContext(finalReaderContext);
-                readerContext.addOnClose(decreasePitContexts);
-                decreasePitContexts = null;
 
                 readerContext.addOnClose(() -> {
+                    openPitContexts.decrementAndGet();
                     searchOperationListener.onFreeReaderContext(finalReaderContext);
                     searchOperationListener.onFreePitContext(finalReaderContext);
                 });
                 // add the newly created pit reader context to active readers
                 putReaderContext(readerContext);
                 readerContext = null;
                 listener.onResponse(finalReaderContext.id());
-                success = true;
             } catch (Exception exc) {
+                Releasables.closeWhileHandlingException(searcherSupplier, readerContext);
                 listener.onFailure(exc);
-            } finally {
-                if (success) {
-                    Releasables.close(readerContext, searcherSupplier, decreasePitContexts);
-                } else {
-                    Releasables.closeWhileHandlingException(searcherSupplier, readerContext, decreasePitContexts);
-                }
             }
         });
     }
@@ -1043,14 +1032,14 @@ public void updatePitIdAndKeepAlive(UpdatePitContextRequest request, ActionListe
         Releasable updatePit = null;
         try {
             updatePit = readerContext.updatePitIdAndKeepAlive(request.getKeepAlive(), request.getPitId(), request.getCreationTime());
-            updatePit.close();
             listener.onResponse(new UpdatePitContextResponse(request.getPitId(), request.getCreationTime(), request.getKeepAlive()));
         } catch (Exception e) {
             freeReaderContext(readerContext.id());
+            listener.onFailure(e);
+        } finally {
             if (updatePit != null) {
                 updatePit.close();
             }
-            listener.onFailure(e);
         }
     }
 

server/src/test/java/org/opensearch/action/search/CreatePitControllerTests.java

@@ -126,7 +126,7 @@ public void onFailure(Exception e) {
         clusterServiceMock = mock(ClusterService.class);
         ClusterState state = mock(ClusterState.class);
 
-        final Settings keepAliveSettings = Settings.builder().put(CreatePitController.PIT_CREATE_PHASE_KEEP_ALIVE.getKey(), 30000).build();
+        final Settings keepAliveSettings = Settings.builder().put(CreatePitController.PIT_INIT_KEEP_ALIVE.getKey(), 30000).build();
         when(clusterServiceMock.getSettings()).thenReturn(keepAliveSettings);
 
         when(state.getMetadata()).thenReturn(Metadata.EMPTY_METADATA);
@@ -177,7 +177,6 @@ public Transport.Connection getConnection(String clusterAlias, DiscoveryNode nod
 
         CreatePitRequest request = new CreatePitRequest(TimeValue.timeValueDays(1), true);
         request.setIndices(new String[] { "index" });
-
         CreatePitController controller = new CreatePitController(
             request,
             searchTransportService,