Merge remote-tracking branch 'origin/main' into upgrade_netty_4.1.126.Final

Author: nicktindall

docs/changelog/134198.yaml

@@ -0,0 +1,5 @@
+pr: 134198
+summary: Improve `ShardLockObtainFailedException` message
+area: Store
+type: enhancement
+issues: []

muted-tests.yml

@@ -507,30 +507,21 @@ tests:
 - class: org.elasticsearch.xpack.ml.integration.ClassificationIT
   method: testWithCustomFeatureProcessors
   issue: https://github.com/elastic/elasticsearch/issues/134001
-- class: org.elasticsearch.cluster.ClusterInfoServiceIT
-  method: testMaxQueueLatenciesInClusterInfo
-  issue: https://github.com/elastic/elasticsearch/issues/134088
 - class: org.elasticsearch.xpack.esql.ccq.MultiClusterSpecIT
   method: test {csv-spec:fork.ForkBeforeStats}
   issue: https://github.com/elastic/elasticsearch/issues/134100
 - 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
 - class: org.elasticsearch.xpack.esql.qa.single_node.GenerativeForkIT
   method: test {csv-spec:inlinestats.MultiIndexInlinestatsOfMultiTypedField}
   issue: https://github.com/elastic/elasticsearch/issues/133973
-- 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:
 #

server/src/internalClusterTest/java/org/elasticsearch/cluster/ClusterInfoServiceIT.java

@@ -27,11 +27,10 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.util.concurrent.TaskExecutionTimeTrackingEsThreadPoolExecutor;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.IndexService;
-import org.elasticsearch.index.shard.GlobalCheckpointListeners;
+import org.elasticsearch.index.seqno.GlobalCheckpointSyncAction;
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.store.Store;
@@ -45,6 +44,7 @@
 import org.elasticsearch.test.InternalTestCluster;
 import org.elasticsearch.test.transport.MockTransportService;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.RequestHandlerRegistry;
 import org.elasticsearch.transport.TransportService;
 import org.hamcrest.Matchers;
 
@@ -57,7 +57,6 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import static java.util.Collections.emptySet;
@@ -431,7 +430,27 @@ public void testMaxQueueLatenciesInClusterInfo() throws Exception {
         final int numShards = randomIntBetween(1, 5);
         createIndex(indexName, Settings.builder().put(SETTING_NUMBER_OF_SHARDS, numShards).put(SETTING_NUMBER_OF_REPLICAS, 0).build());
         ensureGreen(indexName);
-        final var indexService = internalCluster().getInstance(IndicesService.class, dataNodeName).iterator().next();
+
+        // Global checkpoint sync actions are asynchronous. We cannot really tell exactly when they are completely off the
+        // thread pool. To avoid busy waiting, we redirect them to the generic thread pool so that we have precise control
+        // over the write thread pool for assertions.
+        final MockTransportService mockTransportService = MockTransportService.getInstance(dataNodeName);
+        final var originalRegistry = mockTransportService.transport()
+            .getRequestHandlers()
+            .getHandler(GlobalCheckpointSyncAction.ACTION_NAME + "[p]");
+        mockTransportService.transport()
+            .getRequestHandlers()
+            .forceRegister(
+                new RequestHandlerRegistry<>(
+                    GlobalCheckpointSyncAction.ACTION_NAME + "[p]",
+                    in -> null, // no need to deserialize the request since it's local
+                    mockTransportService.getTaskManager(),
+                    originalRegistry.getHandler(),
+                    mockTransportService.getThreadPool().executor(ThreadPool.Names.GENERIC),
+                    true,
+                    true
+                )
+            );
 
         // Block indexing on the data node by submitting write thread pool tasks equal to the number of write threads.
         var barrier = blockDataNodeIndexing(dataNodeName);
@@ -496,28 +515,6 @@ public void testMaxQueueLatenciesInClusterInfo() throws Exception {
             }
             Arrays.stream(threadsToJoin).forEach(thread -> assertFalse(thread.isAlive()));
 
-            // Wait for async post replication actions to complete
-            final var checkpointsSyncLatch = new CountDownLatch(numShards);
-            for (int i = 0; i < numShards; ++i) {
-                final var indexShard = indexService.getShard(i);
-                final long expectedGlobalCheckpoint = indexShard.seqNoStats().getGlobalCheckpoint();
-                logger.info("--> shard [{}] waiting for global checkpoint {}", i, expectedGlobalCheckpoint);
-                indexShard.addGlobalCheckpointListener(expectedGlobalCheckpoint, new GlobalCheckpointListeners.GlobalCheckpointListener() {
-                    @Override
-                    public Executor executor() {
-                        return EsExecutors.DIRECT_EXECUTOR_SERVICE;
-                    }
-
-                    @Override
-                    public void accept(long globalCheckpoint, Exception e) {
-                        assertNull(e); // should have no error
-                        logger.info("--> shard [{}] global checkpoint updated to {}", indexShard.shardId().id(), globalCheckpoint);
-                        checkpointsSyncLatch.countDown();
-                    }
-                }, TimeValue.THIRTY_SECONDS);
-            }
-            safeAwait(checkpointsSyncLatch);
-
             assertThat(
                 "Unexpectedly found a task queued for the write thread pool. Write thread pool dump: " + trackingWriteExecutor,
                 trackingWriteExecutor.peekMaxQueueLatencyInQueueMillis(),

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);

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"));

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());
     }

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();

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)