Fix keep_alive of derived context of point in time (elastic#67398)

The reaper can remove derived contexts of a point in time between 
query and fetch phases as their keep_alive is unset (-1).

Closes elastic#67307

Author: dnhatn

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

@@ -646,18 +646,19 @@ final ReaderContext createOrGetReaderContext(ShardSearchRequest request, boolean
                     searcherSupplier.close();
                     throw e;
                 }
-                return createAndPutReaderContext(request, indexService, shard, searcherSupplier, false);
+                return createAndPutReaderContext(request, indexService, shard, searcherSupplier, false, defaultKeepAlive);
             }
         } else {
+            final long keepAliveInMillis = getKeepAlive(request);
             final IndexService indexService = indicesService.indexServiceSafe(request.shardId().getIndex());
             final IndexShard shard = indexService.getShard(request.shardId().id());
             final Engine.SearcherSupplier searcherSupplier = shard.acquireSearcherSupplier();
-            return createAndPutReaderContext(request, indexService, shard, searcherSupplier, keepStatesInContext);
+            return createAndPutReaderContext(request, indexService, shard, searcherSupplier, keepStatesInContext, keepAliveInMillis);
         }
     }
 
     final ReaderContext createAndPutReaderContext(ShardSearchRequest request, IndexService indexService, IndexShard shard,
-                                                  Engine.SearcherSupplier reader, boolean keepStatesInContext) {
+                                                  Engine.SearcherSupplier reader, boolean keepStatesInContext, long keepAliveInMillis) {
         ReaderContext readerContext = null;
         Releasable decreaseScrollContexts = null;
         try {
@@ -670,16 +671,15 @@ final ReaderContext createAndPutReaderContext(ShardSearchRequest request, IndexS
                             + MAX_OPEN_SCROLL_CONTEXT.getKey() + "] setting.");
                 }
             }
-            final long keepAlive = getKeepAlive(request);
             final ShardSearchContextId id = new ShardSearchContextId(sessionId, idGenerator.incrementAndGet());
             if (keepStatesInContext || request.scroll() != null) {
-                readerContext = new LegacyReaderContext(id, indexService, shard, reader, request, keepAlive);
+                readerContext = new LegacyReaderContext(id, indexService, shard, reader, request, keepAliveInMillis);
                 if (request.scroll() != null) {
                     readerContext.addOnClose(decreaseScrollContexts);
                     decreaseScrollContexts = null;
                 }
             } else {
-                readerContext = new ReaderContext(id, indexService, shard, reader, keepAlive, request.keepAlive() == null);
+                readerContext = new ReaderContext(id, indexService, shard, reader, keepAliveInMillis, true);
             }
             reader = null;
             final ReaderContext finalReaderContext = readerContext;

server/src/test/java/org/elasticsearch/search/SearchServiceTests.java

@@ -583,8 +583,8 @@ public void testMaxOpenScrollContexts() throws Exception {
 
         final ShardScrollRequestTest request = new ShardScrollRequestTest(indexShard.shardId());
         ElasticsearchException ex = expectThrows(ElasticsearchException.class,
-            () -> service.createAndPutReaderContext(
-                request, indexService, indexShard, indexShard.acquireSearcherSupplier(), randomBoolean()));
+            () -> service.createAndPutReaderContext(request, indexService, indexShard, indexShard.acquireSearcherSupplier(),
+                randomBoolean(), SearchService.KEEPALIVE_INTERVAL_SETTING.get(Settings.EMPTY).millis()));
         assertEquals(
             "Trying to create too many scroll contexts. Must be less than or equal to: [" +
                 SearchService.MAX_OPEN_SCROLL_CONTEXT.get(Settings.EMPTY) + "]. " +
@@ -612,8 +612,9 @@ public void testOpenScrollContextsConcurrently() throws Exception {
                     for (; ; ) {
                         final Engine.SearcherSupplier reader = indexShard.acquireSearcherSupplier();
                         try {
-                            searchService.createAndPutReaderContext(
-                                new ShardScrollRequestTest(indexShard.shardId()), indexService, indexShard, reader, true);
+                            final ShardScrollRequestTest request = new ShardScrollRequestTest(indexShard.shardId());
+                            searchService.createAndPutReaderContext(request, indexService, indexShard, reader, true,
+                                SearchService.KEEPALIVE_INTERVAL_SETTING.get(Settings.EMPTY).millis());
                         } catch (ElasticsearchException e) {
                             assertThat(e.getMessage(), equalTo(
                                 "Trying to create too many scroll contexts. Must be less than or equal to: " +
@@ -1120,7 +1121,8 @@ public void testLookUpSearchContext() throws Exception {
                         OriginalIndices.NONE, new SearchRequest().allowPartialSearchResults(true),
                         indexShard.shardId(), 0, 1, new AliasFilter(null, Strings.EMPTY_ARRAY), 1.0f, -1, null);
                     final ReaderContext context = searchService.createAndPutReaderContext(request, indexService, indexShard,
-                        indexShard.acquireSearcherSupplier(), randomBoolean());
+                        indexShard.acquireSearcherSupplier(), randomBoolean(),
+                        SearchService.KEEPALIVE_INTERVAL_SETTING.get(Settings.EMPTY).millis());
                     assertThat(context.id().getId(), equalTo((long) (i + 1)));
                     contextIds.add(context.id());
                 }

x-pack/plugin/frozen-indices/src/internalClusterTest/java/org/elasticsearch/index/engine/FrozenIndexIT.java

@@ -186,7 +186,6 @@ public void testTimestampFieldTypeExposedByAllIndicesServices() throws Exception
         }
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/67307")
     public void testRetryPointInTime() throws Exception {
         internalCluster().ensureAtLeastNumDataNodes(1);
         final List<String> dataNodes =