TMP

Author: kingherc

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/termvectors/30_realtime.yml

@@ -7,7 +7,6 @@
           settings:
             index:
               refresh_interval: -1
-              number_of_replicas: 0
 
  - do:
       cluster.health:

server/src/main/java/org/elasticsearch/action/termvectors/TransportEnsureDocsSearchableAction.java

@@ -17,8 +17,12 @@
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.NoShardAvailableActionException;
+import org.elasticsearch.action.admin.indices.refresh.TransportShardRefreshAction;
 import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.ActiveShardCount;
+import org.elasticsearch.action.support.replication.BasicReplicationRequest;
 import org.elasticsearch.action.support.single.shard.TransportSingleShardAction;
+import org.elasticsearch.client.internal.node.NodeClient;
 import org.elasticsearch.cluster.ProjectState;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.node.DiscoveryNode;
@@ -48,6 +52,7 @@
 public class TransportEnsureDocsSearchableAction extends TransportSingleShardAction<MultiTermVectorsShardRequest, ActionResponse.Empty> {
 
     private static final Logger logger = LogManager.getLogger(TransportEnsureDocsSearchableAction.class);
+    private final NodeClient client;
     private final IndicesService indicesService;
 
     private static final String ACTION_NAME = MultiTermVectorsAction.NAME + "/eds";
@@ -56,6 +61,7 @@ public class TransportEnsureDocsSearchableAction extends TransportSingleShardAct
     @Inject
     public TransportEnsureDocsSearchableAction(
         ClusterService clusterService,
+        NodeClient client,
         TransportService transportService,
         IndicesService indicesService,
         ThreadPool threadPool,
@@ -74,6 +80,7 @@ public TransportEnsureDocsSearchableAction(
             MultiTermVectorsShardRequest::new,
             threadPool.executor(ThreadPool.Names.GET)
         );
+        this.client = client;
         this.indicesService = indicesService;
     }
 
@@ -112,7 +119,7 @@ protected void asyncShardOperation(MultiTermVectorsShardRequest request, ShardId
         assert DiscoveryNode.isStateless(clusterService.getSettings()) : ACTION_NAME + " should only be used in stateless";
         assert DiscoveryNode.hasRole(clusterService.getSettings(), DiscoveryNodeRole.INDEX_ROLE)
             : ACTION_NAME + " should only be executed on a stateless indexing node";
-        logger.debug("received locally {} with {} sub requests", request, request.locations.size());
+        logger.error("received locally {} with {} sub requests", request, request.locations.size());
         getExecutor(shardId).execute(() -> ActionListener.run(listener, l -> {
             final IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
             final IndexShard indexShard = indexService.getShard(shardId.id());
@@ -129,24 +136,45 @@ protected void asyncShardOperation(MultiTermVectorsShardRequest request, ShardId
                     // We prefer simplicity to complexity (trying to avoid the unnecessary stateless refresh) for these limited cases.
                     boolean docInLiveVersionMap = indexShard.withEngine(engine -> engine.isDocumentInLiveVersionMap(docUid));
                     if (docInLiveVersionMap) {
-                        logger.debug("doc id [{}] (uid [{}]) found in live version map of index shard [{}]", docId, docUid, shardId);
+                        logger.error("doc id [{}] (uid [{}]) found in live version map of index shard [{}]", docId, docUid, shardId);
                         docsFoundInLiveVersionMap = true;
                         break;
                     }
                 }
             }
             if (docsFoundInLiveVersionMap) {
-                logger.debug("refreshing index shard [{}] due to mtv_eds", shardId);
-                indexShard.externalRefresh("refresh_mtv_eds", l.map(r -> {
-                    logger.debug("refreshed index shard [{}] due to mtv_eds", shardId);
-                    return ActionResponse.Empty.INSTANCE;
+                logger.error("refreshing index shard [{}] due to mtv_eds", shardId);
+                BasicReplicationRequest refreshRequest = new BasicReplicationRequest(shardId);
+                refreshRequest.waitForActiveShards(ActiveShardCount.NONE);
+                client.executeLocally(TransportShardRefreshAction.TYPE, refreshRequest, l.delegateFailureAndWrap((ll, r) -> {
+                    if (r.getShardInfo().getFailed() > 0) {
+                        throw r.getShardInfo().getFailures()[0];
+                    }
+                    // super.asyncShardOperation(request, shardId, ll);
+                    logger.error("refreshed index shard [{}] due to mtv_eds", shardId);
+                    ll.onResponse(ActionResponse.Empty.INSTANCE);
                 }));
+
+                // indexShard.externalRefresh("refresh_mtv_eds", l.map(r -> {
+                // logger.error("refreshed index shard [{}] due to mtv_eds", shardId);
+                // return ActionResponse.Empty.INSTANCE;
+                // }));
+                // [2025-06-30T13:21:14,574][ERROR][o.e.a.t.TransportTermVectorsAction] [search-GyrIOXh] received locally id 1 for shard
+                // [test_1][2]
+                // [2025-06-30T13:21:14,579][ERROR][o.e.a.t.TransportEnsureDocsSearchableAction] [index-KgxlcuR] received locally
+                // org.elasticsearch.action.termvectors.MultiTermVectorsShardRequest/unset with 1 sub requests
+                // [2025-06-30T13:21:14,581][ERROR][o.e.a.t.TransportEnsureDocsSearchableAction] [index-KgxlcuR] doc id [1] (uid [[fe 1f]])
+                // found in live version map of index shard [[test_1][2]]
+                // [2025-06-30T13:21:14,582][ERROR][o.e.a.t.TransportEnsureDocsSearchableAction] [index-KgxlcuR] refreshing index shard
+                // [[test_1][2]] due to mtv_eds
+                // [2025-06-30T13:21:14,668][ERROR][o.e.a.t.TransportEnsureDocsSearchableAction] [index-KgxlcuR] refreshed index shard
+                // [[test_1][2]] due to mtv_eds
             } else {
                 // Notice that there cannot be a race between the document(s) being evicted from the live version map due to an
                 // ongoing refresh and before the search shards being updated with the new commit, because the documents are
                 // guaranteed to be the in the live version map archive until search shards are updated with the new commit.
                 // Thus, we can safely respond immediately as a no-op.
-                logger.debug("mts_eds does not require refresh of index shard [{}]", shardId);
+                logger.error("mts_eds does not require refresh of index shard [{}]", shardId);
                 l.onResponse(ActionResponse.Empty.INSTANCE);
             }
         }));

server/src/main/java/org/elasticsearch/action/termvectors/TransportShardMultiTermsVectorAction.java

@@ -104,8 +104,10 @@ protected void asyncShardOperation(
         ShardId shardId,
         ActionListener<MultiTermVectorsShardResponse> listener
     ) throws IOException {
-        boolean ensureDocsSearchable = DiscoveryNode.isStateless(clusterService.getSettings())
-            && request.requests.stream().anyMatch(r -> r.realtime() && r.id() != null && r.id().isEmpty() == false);
+        boolean ensureDocsSearchable = DiscoveryNode.isStateless(clusterService.getSettings()) && request.requests.stream().anyMatch(r -> {
+            logger.error("received locally sub-id {} for shard {}", r.id(), shardId);
+            return r.realtime() && r.id() != null && r.id().isEmpty() == false;
+        });
         if (ensureDocsSearchable) {
             client.executeLocally(
                 TransportEnsureDocsSearchableAction.TYPE,
@@ -119,6 +121,7 @@ protected void asyncShardOperation(
 
     @Override
     protected MultiTermVectorsShardResponse shardOperation(MultiTermVectorsShardRequest request, ShardId shardId) {
+        logger.error("servicing locally {} for shard {}", request, shardId);
         final MultiTermVectorsShardResponse response = new MultiTermVectorsShardResponse();
         final IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
         final IndexShard indexShard = indexService.getShard(shardId.id());

server/src/main/java/org/elasticsearch/action/termvectors/TransportTermVectorsAction.java

@@ -110,6 +110,7 @@ protected void asyncShardOperation(TermVectorsRequest request, ShardId shardId,
             boolean ensureDocSearchable = DiscoveryNode.isStateless(clusterService.getSettings())
                 && request.id() != null
                 && request.id().isEmpty() == false;
+            logger.error("received locally id {} for shard {}", request.id(), shardId);
             if (ensureDocSearchable) {
                 // Ensure that the document is searchable before we execute the term vectors request
                 MultiTermVectorsShardRequest ensureDocSearchableRequest = new MultiTermVectorsShardRequest(request.index(), shardId.id());
@@ -137,6 +138,7 @@ protected void asyncShardOperation(TermVectorsRequest request, ShardId shardId,
 
     @Override
     protected TermVectorsResponse shardOperation(TermVectorsRequest request, ShardId shardId) {
+        logger.error("servicing locally id {} for shard {}", request.id(), shardId);
         IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
         IndexShard indexShard = indexService.getShard(shardId.id());
         return TermVectorsService.getTermVectors(indexShard, request);