Merge branch 'main' into es-134088-fix

elasticmachine · web-flow · commit fc74a9965328 · 2025-09-05T05:12:44.000-04:00
diff --git a/docs/changelog/134198.yaml b/docs/changelog/134198.yaml
@@ -0,0 +1,5 @@
+pr: 134198
+summary: Improve `ShardLockObtainFailedException` message
+area: Store
+type: enhancement
+issues: []
diff --git a/muted-tests.yml b/muted-tests.yml
@@ -513,12 +513,6 @@ tests:
 - class: org.elasticsearch.xpack.esql.ccq.MultiClusterSpecIT
   method: test {csv-spec:spatial.ConvertFromStringParseError}
   issue: https://github.com/elastic/elasticsearch/issues/134104
-- class: org.elasticsearch.xpack.writeloadforecaster.WriteLoadForecasterIT
-  method: testWriteLoadForecastIsOverriddenBySetting
-  issue: https://github.com/elastic/elasticsearch/issues/133455
-- class: org.elasticsearch.xpack.writeloadforecaster.WriteLoadForecasterIT
-  method: testWriteLoadForecastDoesNotGetPopulatedWithInvalidLicense
-  issue: https://github.com/elastic/elasticsearch/issues/134124
 - class: org.elasticsearch.xpack.esql.ccq.MultiClusterSpecIT
   method: test {csv-spec:fork.ForkWithMixOfCommands}
   issue: https://github.com/elastic/elasticsearch/issues/134135
@@ -528,6 +522,9 @@ tests:
 - class: org.elasticsearch.test.rest.yaml.CcsCommonYamlTestSuiteIT
   method: test {p0=search/510_range_query_out_of_bounds/Test range query for float field with out of bounds lower limit}
   issue: https://github.com/elastic/elasticsearch/issues/134184
+- class: org.elasticsearch.health.HealthPeriodicLoggerTests
+  method: testOutputModeNoLogging
+  issue: https://github.com/elastic/elasticsearch/issues/134200
 
 # Examples:
 #
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/indices.cancel_migrate_reindex.json b/rest-api-spec/src/main/resources/rest-api-spec/api/indices.cancel_migrate_reindex.json
@@ -4,8 +4,8 @@
       "url":"https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-indices-cancel-migrate-reindex",
       "description":"Cancel a migration reindex operation"
     },
-    "stability":"experimental",
-    "visibility":"private",
+    "stability":"stable",
+    "visibility":"public",
     "headers":{
       "accept": [ "application/json"],
       "content_type": ["application/json"]
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/indices.create_from.json b/rest-api-spec/src/main/resources/rest-api-spec/api/indices.create_from.json
@@ -4,8 +4,8 @@
       "url":"https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-indices-create-from",
       "description":"Create an index from a source index"
     },
-    "stability":"experimental",
-    "visibility":"private",
+    "stability":"stable",
+    "visibility":"public",
     "headers":{
       "accept": [ "application/json"],
       "content_type": ["application/json"]
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/indices.get_migrate_reindex_status.json b/rest-api-spec/src/main/resources/rest-api-spec/api/indices.get_migrate_reindex_status.json
@@ -4,8 +4,8 @@
       "url":"https://www.elastic.co/docs/api/doc/elasticsearch/group/endpoint-migration",
       "description":"Get the migration reindexing status"
     },
-    "stability":"experimental",
-    "visibility":"private",
+    "stability":"stable",
+    "visibility":"public",
     "headers":{
       "accept": [ "application/json"],
       "content_type": ["application/json"]
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/indices.migrate_reindex.json b/rest-api-spec/src/main/resources/rest-api-spec/api/indices.migrate_reindex.json
@@ -4,8 +4,8 @@
       "url":"https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-indices-migrate-reindex",
       "description":"Reindex legacy backing indices"
     },
-    "stability":"experimental",
-    "visibility":"private",
+    "stability":"stable",
+    "visibility":"public",
     "headers":{
       "accept": [ "application/json"],
       "content_type": ["application/json"]
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/query_rules.test.json b/rest-api-spec/src/main/resources/rest-api-spec/api/query_rules.test.json
@@ -4,7 +4,7 @@
       "url": "https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-query-rules-test",
       "description": "Test a query ruleset"
     },
-    "stability": "experimental",
+    "stability": "stable",
     "visibility": "public",
     "headers": {
       "accept": [
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/security.activate_user_profile.json b/rest-api-spec/src/main/resources/rest-api-spec/api/security.activate_user_profile.json
@@ -5,7 +5,7 @@
       "description":"Activate a user profile"
     },
     "stability":"stable",
-    "visibility":"private",
+    "visibility":"public",
     "headers":{
       "accept": [ "application/json"],
       "content_type": ["application/json"]
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/security.disable_user_profile.json b/rest-api-spec/src/main/resources/rest-api-spec/api/security.disable_user_profile.json
@@ -5,7 +5,7 @@
       "description":"Disable a user profile"
     },
     "stability":"stable",
-    "visibility":"private",
+    "visibility":"public",
     "headers":{
       "accept": [ "application/json"]
     },
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/security.enable_user_profile.json b/rest-api-spec/src/main/resources/rest-api-spec/api/security.enable_user_profile.json
@@ -5,7 +5,7 @@
       "description":"Enable a user profile"
     },
     "stability":"stable",
-    "visibility":"private",
+    "visibility":"public",
     "headers":{
       "accept": [ "application/json"]
     },
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/security.get_user_profile.json b/rest-api-spec/src/main/resources/rest-api-spec/api/security.get_user_profile.json
@@ -5,7 +5,7 @@
       "description":"Get a user profile"
     },
     "stability":"stable",
-    "visibility":"private",
+    "visibility":"public",
     "headers":{
       "accept": [ "application/json"]
     },
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/security.has_privileges_user_profile.json b/rest-api-spec/src/main/resources/rest-api-spec/api/security.has_privileges_user_profile.json
@@ -5,7 +5,7 @@
       "description":"Check user profile privileges"
     },
     "stability":"stable",
-    "visibility":"private",
+    "visibility":"public",
     "headers":{
       "accept": [ "application/json"],
       "content_type": ["application/json"]
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/security.suggest_user_profiles.json b/rest-api-spec/src/main/resources/rest-api-spec/api/security.suggest_user_profiles.json
@@ -5,7 +5,7 @@
       "description":"Suggest a user profile"
     },
     "stability":"stable",
-    "visibility":"private",
+    "visibility":"public",
     "headers":{
       "accept": [ "application/json"],
       "content_type": ["application/json"]
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/security.update_user_profile_data.json b/rest-api-spec/src/main/resources/rest-api-spec/api/security.update_user_profile_data.json
@@ -5,7 +5,7 @@
       "description":"Update user profile data"
     },
     "stability":"stable",
-    "visibility":"private",
+    "visibility":"public",
     "headers":{
       "accept": [ "application/json"],
       "content_type": ["application/json"]
diff --git a/server/src/main/java/org/elasticsearch/env/NodeEnvironment.java b/server/src/main/java/org/elasticsearch/env/NodeEnvironment.java
@@ -1009,12 +1009,16 @@ void acquire(long timeoutInMillis, final String details) throws ShardLockObtainF
                     setDetails(details);
                 } else {
                     final Tuple<Long, String> lockDetails = this.lockDetails; // single volatile read
+                    final var stateAge = TimeValue.timeValueNanos(System.nanoTime() - lockDetails.v1());
                     final var message = format(
-                        "obtaining shard lock for [%s] timed out after [%dms], lock already held for [%s] with age [%dms]",
+                        """
+                            obtaining shard lock for [%s] timed out after [%dms]; \
+                            this shard lock is still held by a different instance of the shard and has been in state [%s] for [%s/%dms]""",
                         details,
                         timeoutInMillis,
                         lockDetails.v2(),
-                        TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - lockDetails.v1())
+                        stateAge,
+                        stateAge.millis()
                     );
                     maybeLogThreadDump(shardId, message);
                     throw new ShardLockObtainFailedException(shardId, message);
diff --git a/server/src/test/java/org/elasticsearch/env/NodeEnvironmentTests.java b/server/src/test/java/org/elasticsearch/env/NodeEnvironmentTests.java
@@ -70,6 +70,7 @@
 import static org.hamcrest.Matchers.arrayWithSize;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.matchesPattern;
 import static org.hamcrest.Matchers.startsWith;
 
 @LuceneTestCase.SuppressFileSystems("ExtrasFS") // TODO: fix test to allow extras
@@ -134,25 +135,28 @@ public void testShardLock() throws Exception {
 
             try (var mockLog = MockLog.capture(NodeEnvironment.class); var lock = env.shardLock(new ShardId(index, 0), "1")) {
                 mockLog.addExpectation(
-                    new MockLog.SeenEventExpectation(
-                        "hot threads logging",
-                        NODE_ENVIRONMENT_LOGGER_NAME,
-                        Level.DEBUG,
-                        "hot threads while failing to obtain shard lock for [foo][0]: obtaining shard lock for [2] timed out after *"
-                    )
+                    new MockLog.SeenEventExpectation("hot threads logging", NODE_ENVIRONMENT_LOGGER_NAME, Level.DEBUG, """
+                        hot threads while failing to obtain shard lock for [foo][0]: obtaining shard lock for [2] timed out after [*ms]; \
+                        this shard lock is still held by a different instance of the shard and has been in state [1] for [*/*ms]*""")
                 );
                 mockLog.addExpectation(
                     new MockLog.UnseenEventExpectation(
                         "second attempt should be suppressed due to throttling",
                         NODE_ENVIRONMENT_LOGGER_NAME,
                         Level.DEBUG,
-                        "hot threads while failing to obtain shard lock for [foo][0]: obtaining shard lock for [3] timed out after *"
+                        "*obtaining shard lock for [3] timed out*"
                     )
                 );
 
                 assertEquals(new ShardId(index, 0), lock.getShardId());
 
-                expectThrows(ShardLockObtainFailedException.class, () -> env.shardLock(new ShardId(index, 0), "2"));
+                assertThat(
+                    expectThrows(ShardLockObtainFailedException.class, () -> env.shardLock(new ShardId(index, 0), "2")).getMessage(),
+                    matchesPattern("""
+                        \\[foo]\\[0]: obtaining shard lock for \\[2] timed out after \\[0ms]; \
+                        this shard lock is still held by a different instance of the shard \
+                        and has been in state \\[1] for \\[.*/[0-9]+ms]""")
+                );
 
                 for (Path path : env.indexPaths(index)) {
                     Files.createDirectories(path.resolve("0"));
diff --git a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/AbstractEnrichBasedCrossClusterTestCase.java b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/AbstractEnrichBasedCrossClusterTestCase.java
@@ -87,6 +87,7 @@ protected Settings nodeSettings() {
     }
 
     static final EnrichPolicy hostPolicy = new EnrichPolicy("match", null, List.of("hosts"), "ip", List.of("ip", "os"));
+    static final EnrichPolicy hostPolicyLocal = new EnrichPolicy("match", null, List.of("hosts_local"), "ip", List.of("ip", "os"));
     static final EnrichPolicy vendorPolicy = new EnrichPolicy("match", null, List.of("vendors"), "os", List.of("os", "vendor"));
 
     @Before
@@ -115,18 +116,23 @@ public void setupHostsEnrich() {
             "Windows"
         );
         for (String cluster : allClusters()) {
-            Client client = client(cluster);
-            client.admin().indices().prepareCreate("hosts").setMapping("ip", "type=ip", "os", "type=keyword").get();
-            for (Map.Entry<String, String> h : allHosts.entrySet()) {
-                client.prepareIndex("hosts").setSource("ip", h.getKey(), "os", h.getValue()).get();
-            }
-            client.admin().indices().prepareRefresh("hosts").get();
-            client.execute(PutEnrichPolicyAction.INSTANCE, new PutEnrichPolicyAction.Request(TEST_REQUEST_TIMEOUT, "hosts", hostPolicy))
-                .actionGet();
-            client.execute(ExecuteEnrichPolicyAction.INSTANCE, new ExecuteEnrichPolicyAction.Request(TEST_REQUEST_TIMEOUT, "hosts"))
-                .actionGet();
-            assertAcked(client.admin().indices().prepareDelete("hosts"));
+            initHostsPolicy(client(cluster), "hosts", hostPolicy, allHosts);
+        }
+        // create policy on coordinator only
+        initHostsPolicy(client(), "hosts_local", hostPolicyLocal, allHosts);
+    }
+
+    private static void initHostsPolicy(Client client, String indexName, EnrichPolicy policy, Map<String, String> allHosts) {
+        client.admin().indices().prepareCreate(indexName).setMapping("ip", "type=ip", "os", "type=keyword").get();
+        for (Map.Entry<String, String> h : allHosts.entrySet()) {
+            client.prepareIndex(indexName).setSource("ip", h.getKey(), "os", h.getValue()).get();
         }
+        client.admin().indices().prepareRefresh(indexName).get();
+        client.execute(PutEnrichPolicyAction.INSTANCE, new PutEnrichPolicyAction.Request(TEST_REQUEST_TIMEOUT, indexName, policy))
+            .actionGet();
+        client.execute(ExecuteEnrichPolicyAction.INSTANCE, new ExecuteEnrichPolicyAction.Request(TEST_REQUEST_TIMEOUT, indexName))
+            .actionGet();
+        assertAcked(client.admin().indices().prepareDelete(indexName));
     }
 
     @Before
@@ -201,7 +207,7 @@ record Event(long timestamp, String user, String host) {
     public void wipeEnrichPolicies() {
         for (String cluster : allClusters()) {
             cluster(cluster).wipe(Set.of());
-            for (String policy : List.of("hosts", "vendors")) {
+            for (String policy : List.of("hosts", "hosts_local", "vendors")) {
                 if (tolerateErrorsWhenWipingEnrichPolicies()) {
                     try {
                         client(cluster).execute(
@@ -226,6 +232,10 @@ static String enrichHosts(Enrich.Mode mode) {
         return EsqlTestUtils.randomEnrichCommand("hosts", mode, hostPolicy.getMatchField(), hostPolicy.getEnrichFields());
     }
 
+    static String enrichHostsLocal(Enrich.Mode mode) {
+        return EsqlTestUtils.randomEnrichCommand("hosts_local", mode, hostPolicyLocal.getMatchField(), hostPolicyLocal.getEnrichFields());
+    }
+
     static String enrichVendors(Enrich.Mode mode) {
         return EsqlTestUtils.randomEnrichCommand("vendors", mode, vendorPolicy.getMatchField(), vendorPolicy.getEnrichFields());
     }
diff --git a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterEnrichIT.java b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterEnrichIT.java
@@ -110,6 +110,52 @@ public void testWithHostsPolicy() {
         }
     }
 
+    public void testFromRemotesWithCoordPolicy() {
+
+        String query = "FROM *:events | eval ip= TO_STR(host) | "
+            + enrichHostsLocal(Enrich.Mode.COORDINATOR)
+            + " | stats c = COUNT(*) by os | SORT os";
+        try (EsqlQueryResponse resp = runQuery(query, null)) {
+            List<List<Object>> rows = getValuesList(resp);
+            assertThat(
+                rows,
+                equalTo(
+                    List.of(
+                        List.of(1L, "Android"),
+                        List.of(2L, "Linux"),
+                        List.of(4L, "MacOS"),
+                        List.of(3L, "Windows"),
+                        List.of(1L, "iOS"),
+                        Arrays.asList(2L, (String) null)
+                    )
+                )
+            );
+            assertTrue(resp.getExecutionInfo().isCrossClusterSearch());
+        }
+
+        query = "FROM *:events | eval ip= TO_STR(host) | stats by ip | "
+            + enrichHostsLocal(Enrich.Mode.COORDINATOR)
+            + " | stats c = COUNT(*) by os | SORT os";
+        try (EsqlQueryResponse resp = runQuery(query, null)) {
+            List<List<Object>> rows = getValuesList(resp);
+            assertThat(
+                rows,
+                equalTo(
+                    List.of(
+                        List.of(1L, "Android"),
+                        List.of(2L, "Linux"),
+                        List.of(2L, "MacOS"),
+                        List.of(2L, "Windows"),
+                        List.of(1L, "iOS"),
+                        Arrays.asList(2L, (String) null)
+                    )
+                )
+            );
+            assertTrue(resp.getExecutionInfo().isCrossClusterSearch());
+        }
+
+    }
+
     public void testEnrichHostsAggThenEnrichVendorCoordinator() {
         Tuple<Boolean, Boolean> includeCCSMetadata = randomIncludeCCSMetadata();
         Boolean requestIncludeMeta = includeCCSMetadata.v1();
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlSession.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlSession.java
@@ -78,6 +78,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -372,22 +373,34 @@ public void analyzedPlan(
             return;
         }
 
-        PreAnalyzer.PreAnalysis preAnalysis = preAnalyzer.preAnalyze(parsed);
+        var preAnalysis = preAnalyzer.preAnalyze(parsed);
         EsqlCCSUtils.initCrossClusterState(indicesExpressionGrouper, verifier.licenseState(), preAnalysis.indices, executionInfo);
 
-        var listener = SubscribableListener. //
+        SubscribableListener. //
         <EnrichResolution>newForked(l -> enrichPolicyResolver.resolvePolicies(preAnalysis.enriches, executionInfo, l))
             .<PreAnalysisResult>andThenApply(enrichResolution -> FieldNameUtils.resolveFieldNames(parsed, enrichResolution))
-            .<PreAnalysisResult>andThen((l, preAnalysisResult) -> resolveInferences(parsed, preAnalysisResult, l));
-        // first resolve the lookup indices, then the main indices
-        for (var index : preAnalysis.lookupIndices) {
-            listener = listener.andThen((l, preAnalysisResult) -> preAnalyzeLookupIndex(index, preAnalysisResult, executionInfo, l));
-        }
-        listener.<PreAnalysisResult>andThen((l, result) -> preAnalyzeMainIndices(preAnalysis, executionInfo, result, requestFilter, l))
-            .<LogicalPlan>andThen((l, result) -> analyzeWithRetry(parsed, requestFilter, preAnalysis, executionInfo, result, l))
+            .<PreAnalysisResult>andThen((l, r) -> resolveInferences(parsed, r, l))
+            .<PreAnalysisResult>andThen((l, r) -> preAnalyzeMainIndices(preAnalysis, executionInfo, r, requestFilter, l))
+            .<PreAnalysisResult>andThen((l, r) -> preAnalyzeLookupIndices(preAnalysis.lookupIndices.iterator(), r, executionInfo, l))
+            .<LogicalPlan>andThen((l, r) -> analyzeWithRetry(parsed, requestFilter, preAnalysis, executionInfo, r, l))
             .addListener(logicalPlanListener);
     }
 
+    private void preAnalyzeLookupIndices(
+        Iterator<IndexPattern> lookupIndices,
+        PreAnalysisResult preAnalysisResult,
+        EsqlExecutionInfo executionInfo,
+        ActionListener<PreAnalysisResult> listener
+    ) {
+        if (lookupIndices.hasNext()) {
+            preAnalyzeLookupIndex(lookupIndices.next(), preAnalysisResult, executionInfo, listener.delegateFailureAndWrap((l, r) -> {
+                preAnalyzeLookupIndices(lookupIndices, r, executionInfo, l);
+            }));
+        } else {
+            listener.onResponse(preAnalysisResult);
+        }
+    }
+
     private void preAnalyzeLookupIndex(
         IndexPattern lookupIndexPattern,
         PreAnalysisResult result,
diff --git a/x-pack/plugin/write-load-forecaster/src/internalClusterTest/java/org/elasticsearch/xpack/writeloadforecaster/WriteLoadForecasterIT.java b/x-pack/plugin/write-load-forecaster/src/internalClusterTest/java/org/elasticsearch/xpack/writeloadforecaster/WriteLoadForecasterIT.java
@@ -145,6 +145,8 @@ private void setUpDataStreamWriteDocsAndRollover(String dataStreamName) throws E
     private void setUpDataStreamWriteDocsAndRollover(String dataStreamName, Settings extraIndexTemplateSettings) throws Exception {
         final int numberOfShards = randomIntBetween(1, 5);
         final int numberOfReplicas = randomIntBetween(0, 1);
+        internalCluster().ensureAtLeastNumDataNodes(numberOfReplicas + 1);
+
         final Settings indexSettings = Settings.builder()
             .put(extraIndexTemplateSettings)
             .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, numberOfShards)
