Merge remote-tracking branch 'upstream/main' into fix_120874

jordan-powers · jordan-powers · commit 3c669c072945 · 2025-02-10T18:05:51.000-08:00
diff --git a/muted-tests.yml b/muted-tests.yml
@@ -361,18 +361,13 @@ tests:
 - class: org.elasticsearch.analysis.common.CommonAnalysisClientYamlTestSuiteIT
   method: test {yaml=analysis-common/40_token_filters/stemmer_override file access}
   issue: https://github.com/elastic/elasticsearch/issues/121625
-- class: org.elasticsearch.xpack.esql.plugin.DataNodeRequestSenderTests
-  method: testDoNotRetryOnRequestLevelFailure
-  issue: https://github.com/elastic/elasticsearch/issues/121966
 - class: org.elasticsearch.xpack.searchablesnapshots.hdfs.SecureHdfsSearchableSnapshotsIT
   issue: https://github.com/elastic/elasticsearch/issues/121967
 - class: org.elasticsearch.action.search.SearchQueryThenFetchAsyncActionTests
   method: testBottomFieldSort
   issue: https://github.com/elastic/elasticsearch/issues/121503
 - class: org.elasticsearch.xpack.application.CohereServiceUpgradeIT
   issue: https://github.com/elastic/elasticsearch/issues/121537
-- class: org.elasticsearch.xpack.migrate.action.ReindexDatastreamIndexTransportActionIT
-  issue: https://github.com/elastic/elasticsearch/issues/121737
 - class: org.elasticsearch.xpack.restart.FullClusterRestartIT
   method: testWatcherWithApiKey {cluster=UPGRADED}
   issue: https://github.com/elastic/elasticsearch/issues/122061
@@ -388,16 +383,8 @@ tests:
 - class: org.elasticsearch.xpack.ml.integration.ClassificationIT
   method: testWithOnlyTrainingRowsAndTrainingPercentIsFifty_DependentVariableIsBoolean
   issue: https://github.com/elastic/elasticsearch/issues/121680
-- class: org.elasticsearch.xpack.esql.action.EsqlActionBreakerIT
-  issue: https://github.com/elastic/elasticsearch/issues/122153
 - class: org.elasticsearch.xpack.esql.qa.single_node.EsqlSpecIT
   issue: https://github.com/elastic/elasticsearch/issues/121411
-- class: org.elasticsearch.xpack.esql.action.EsqlNodeFailureIT
-  method: testFailureLoadingFields
-  issue: https://github.com/elastic/elasticsearch/issues/122132
-- class: org.elasticsearch.xpack.esql.action.CrossClustersCancellationIT
-  method: testCloseSkipUnavailable
-  issue: https://github.com/elastic/elasticsearch/issues/121627
 - class: org.elasticsearch.xpack.downsample.DownsampleActionSingleNodeTests
   method: testDuplicateDownsampleRequest
   issue: https://github.com/elastic/elasticsearch/issues/122158
@@ -413,6 +400,17 @@ tests:
 - class: org.elasticsearch.xpack.esql.analysis.VerifierTests
   method: testChangePoint_valueNumeric
   issue: https://github.com/elastic/elasticsearch/issues/122181
+- class: org.elasticsearch.datastreams.TSDBPassthroughIndexingIT
+  issue: https://github.com/elastic/elasticsearch/issues/121716
+- class: org.elasticsearch.smoketest.SmokeTestMonitoringWithSecurityIT
+  method: testHTTPExporterWithSSL
+  issue: https://github.com/elastic/elasticsearch/issues/122220
+- class: org.elasticsearch.xpack.esql.action.CrossClusterAsyncQueryStopIT
+  method: testStopQueryLocal
+  issue: https://github.com/elastic/elasticsearch/issues/121672
+- class: org.elasticsearch.xpack.security.authz.IndexAliasesTests
+  method: testRemoveIndex
+  issue: https://github.com/elastic/elasticsearch/issues/122221
 
 # Examples:
 #
diff --git a/server/src/test/java/org/elasticsearch/bootstrap/NoSecurityManagerTests.java b/server/src/test/java/org/elasticsearch/bootstrap/NoSecurityManagerTests.java
@@ -12,6 +12,7 @@
 import com.carrotsearch.randomizedtesting.annotations.ThreadLeakFilters;
 
 import org.apache.lucene.tests.util.LuceneTestCase;
+import org.elasticsearch.jdk.RuntimeVersionFeature;
 import org.elasticsearch.test.GraalVMThreadsFilter;
 
 import static org.hamcrest.Matchers.is;
@@ -20,6 +21,7 @@
 public class NoSecurityManagerTests extends LuceneTestCase {
 
     public void testPrepopulateSecurityCaller() {
+        assumeTrue("security manager must be available", RuntimeVersionFeature.isSecurityManagerAvailable());
         assumeTrue("Unexpected security manager:" + System.getSecurityManager(), System.getSecurityManager() == null);
         boolean isAtLeastJava17 = Runtime.version().feature() >= 17;
         boolean isPrepopulated = Security.prepopulateSecurityCaller();
diff --git a/server/src/test/java/org/elasticsearch/plugins/PluginsServiceTests.java b/server/src/test/java/org/elasticsearch/plugins/PluginsServiceTests.java
@@ -18,6 +18,7 @@
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.TestEnvironment;
 import org.elasticsearch.index.IndexModule;
+import org.elasticsearch.jdk.RuntimeVersionFeature;
 import org.elasticsearch.plugin.analysis.CharFilterFactory;
 import org.elasticsearch.plugins.scanners.PluginInfo;
 import org.elasticsearch.plugins.spi.BarPlugin;
@@ -873,6 +874,7 @@ public Reader create(Reader reader) {
     }
 
     public void testCanCreateAClassLoader() {
+        assumeTrue("security manager must be available", RuntimeVersionFeature.isSecurityManagerAvailable());
         assertEquals(
             "access denied (\"java.lang.RuntimePermission\" \"createClassLoader\")",
             expectThrows(AccessControlException.class, () -> new Loader(this.getClass().getClassLoader())).getMessage()
diff --git a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/AbstractCrossClusterUsageTelemetryIT.java b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/AbstractCrossClusterUsageTelemetryIT.java
@@ -33,8 +33,8 @@
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertResponse;
 
-public class AbstractCrossClustersUsageTelemetryIT extends AbstractMultiClustersTestCase {
-    private static final Logger LOGGER = LogManager.getLogger(AbstractCrossClustersUsageTelemetryIT.class);
+public class AbstractCrossClusterUsageTelemetryIT extends AbstractMultiClustersTestCase {
+    private static final Logger LOGGER = LogManager.getLogger(AbstractCrossClusterUsageTelemetryIT.class);
     protected static final String REMOTE1 = "cluster-a";
     protected static final String REMOTE2 = "cluster-b";
     protected static final String LOCAL_INDEX = "logs-1";
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
@@ -75,7 +75,7 @@ protected Collection<String> allClusters() {
     @Override
     protected Collection<Class<? extends Plugin>> nodePlugins(String clusterAlias) {
         List<Class<? extends Plugin>> plugins = new ArrayList<>(super.nodePlugins(clusterAlias));
-        plugins.add(CrossClustersEnrichIT.LocalStateEnrich.class);
+        plugins.add(CrossClusterEnrichIT.LocalStateEnrich.class);
         plugins.add(IngestCommonPlugin.class);
         plugins.add(ReindexPlugin.class);
         return plugins;
@@ -274,7 +274,7 @@ protected List<ActionType<XPackInfoFeatureResponse>> infoActions() {
 
         @Override
         protected Class<? extends TransportAction<XPackInfoRequest, XPackInfoResponse>> getInfoAction() {
-            return CrossClustersQueriesWithInvalidLicenseIT.LocalStateEnrich.EnrichTransportXPackInfoAction.class;
+            return CrossClusterQueriesWithInvalidLicenseIT.LocalStateEnrich.EnrichTransportXPackInfoAction.class;
         }
     }
 }
diff --git a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterCancellationIT.java b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterCancellationIT.java
@@ -44,7 +44,7 @@
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.instanceOf;
 
-public class CrossClustersCancellationIT extends AbstractMultiClustersTestCase {
+public class CrossClusterCancellationIT extends AbstractMultiClustersTestCase {
     private static final String REMOTE_CLUSTER = "cluster-a";
 
     @Override
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
@@ -27,7 +27,7 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 
-public class CrossClustersEnrichIT extends AbstractEnrichBasedCrossClusterTestCase {
+public class CrossClusterEnrichIT extends AbstractEnrichBasedCrossClusterTestCase {
 
     @Override
     protected Collection<Class<? extends Plugin>> nodePlugins(String clusterAlias) {
diff --git a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterQueriesWithInvalidLicenseIT.java b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterQueriesWithInvalidLicenseIT.java
@@ -27,7 +27,7 @@
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.is;
 
-public class CrossClustersQueriesWithInvalidLicenseIT extends AbstractEnrichBasedCrossClusterTestCase {
+public class CrossClusterQueriesWithInvalidLicenseIT extends AbstractEnrichBasedCrossClusterTestCase {
 
     private static final String LICENSE_ERROR_MESSAGE = "A valid Enterprise license is required to run ES|QL cross-cluster searches.";
 
diff --git a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterUsageTelemetryIT.java b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterUsageTelemetryIT.java
@@ -34,7 +34,7 @@
 import static org.elasticsearch.xpack.esql.action.EsqlAsyncTestUtils.deleteAsyncId;
 import static org.hamcrest.Matchers.equalTo;
 
-public class CrossClustersUsageTelemetryIT extends AbstractCrossClustersUsageTelemetryIT {
+public class CrossClusterUsageTelemetryIT extends AbstractCrossClusterUsageTelemetryIT {
     private static final String INDEX_WITH_RUNTIME_MAPPING = "blocking";
 
     @Override
diff --git a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterUsageTelemetryNoLicenseIT.java b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterUsageTelemetryNoLicenseIT.java
@@ -17,7 +17,7 @@
 
 import static org.hamcrest.Matchers.equalTo;
 
-public class CrossClustersUsageTelemetryNoLicenseIT extends AbstractCrossClustersUsageTelemetryIT {
+public class CrossClusterUsageTelemetryNoLicenseIT extends AbstractCrossClusterUsageTelemetryIT {
 
     @Override
     protected Collection<Class<? extends Plugin>> nodePlugins(String clusterAlias) {
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/DataNodeRequestSender.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/DataNodeRequestSender.java
@@ -34,7 +34,9 @@
 import org.elasticsearch.xpack.esql.action.EsqlSearchShardsAction;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.IdentityHashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -58,6 +60,7 @@ abstract class DataNodeRequestSender {
     private final Map<DiscoveryNode, Semaphore> nodePermits = new HashMap<>();
     private final Map<ShardId, ShardFailure> shardFailures = ConcurrentCollections.newConcurrentMap();
     private final AtomicBoolean changed = new AtomicBoolean();
+    private boolean reportedFailure = false; // guarded by sendingLock
 
     DataNodeRequestSender(TransportService transportService, Executor esqlExecutor, CancellableTask rootTask) {
         this.transportService = transportService;
@@ -117,11 +120,14 @@ private void trySendingRequestsForPendingShards(TargetShards targetShards, Compu
                             );
                         }
                     }
-                    if (shardFailures.values().stream().anyMatch(shardFailure -> shardFailure.fatal)) {
-                        for (var e : shardFailures.values()) {
-                            computeListener.acquireAvoid().onFailure(e.failure);
-                        }
+                    if (reportedFailure || shardFailures.values().stream().anyMatch(shardFailure -> shardFailure.fatal)) {
+                        reportedFailure = true;
+                        reportFailures(computeListener);
                     } else {
+                        pendingShardIds.removeIf(shr -> {
+                            var failure = shardFailures.get(shr);
+                            return failure != null && failure.fatal;
+                        });
                         var nodeRequests = selectNodeRequests(targetShards);
                         for (NodeRequest request : nodeRequests) {
                             sendOneNodeRequest(targetShards, computeListener, request);
@@ -136,6 +142,20 @@ private void trySendingRequestsForPendingShards(TargetShards targetShards, Compu
         }
     }
 
+    private void reportFailures(ComputeListener computeListener) {
+        assert sendingLock.isHeldByCurrentThread();
+        assert reportedFailure;
+        Iterator<ShardFailure> it = shardFailures.values().iterator();
+        Set<Exception> seen = Collections.newSetFromMap(new IdentityHashMap<>());
+        while (it.hasNext()) {
+            ShardFailure failure = it.next();
+            if (seen.add(failure.failure)) {
+                computeListener.acquireAvoid().onFailure(failure.failure);
+            }
+            it.remove();
+        }
+    }
+
     private void sendOneNodeRequest(TargetShards targetShards, ComputeListener computeListener, NodeRequest request) {
         final ActionListener<List<DriverProfile>> listener = computeListener.acquireCompute();
         sendRequest(request.node, request.shardIds, request.aliasFilters, new NodeListener() {
@@ -148,7 +168,7 @@ void onAfter(List<DriverProfile> profiles) {
             @Override
             public void onResponse(DataNodeComputeResponse response) {
                 // remove failures of successful shards
-                for (ShardId shardId : targetShards.shardIds()) {
+                for (ShardId shardId : request.shardIds()) {
                     if (response.shardLevelFailures().containsKey(shardId) == false) {
                         shardFailures.remove(shardId);
                     }
@@ -250,6 +270,7 @@ private List<NodeRequest> selectNodeRequests(TargetShards targetShards) {
         final Iterator<ShardId> shardsIt = pendingShardIds.iterator();
         while (shardsIt.hasNext()) {
             ShardId shardId = shardsIt.next();
+            assert shardFailures.get(shardId) == null || shardFailures.get(shardId).fatal == false;
             TargetShard shard = targetShards.getShard(shardId);
             Iterator<DiscoveryNode> nodesIt = shard.remainingNodes.iterator();
             DiscoveryNode selectedNode = null;
diff --git a/x-pack/plugin/migrate/src/internalClusterTest/java/org/elasticsearch/xpack/migrate/action/ReindexDatastreamIndexTransportActionIT.java b/x-pack/plugin/migrate/src/internalClusterTest/java/org/elasticsearch/xpack/migrate/action/ReindexDatastreamIndexTransportActionIT.java
@@ -24,8 +24,6 @@
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.bulk.BulkResponse;
 import org.elasticsearch.action.index.IndexRequest;
-import org.elasticsearch.action.ingest.DeletePipelineRequest;
-import org.elasticsearch.action.ingest.DeletePipelineTransportAction;
 import org.elasticsearch.action.ingest.PutPipelineRequest;
 import org.elasticsearch.action.ingest.PutPipelineTransportAction;
 import org.elasticsearch.cluster.block.ClusterBlockException;
@@ -70,18 +68,8 @@
 
 public class ReindexDatastreamIndexTransportActionIT extends ESIntegTestCase {
     @After
-    private void cleanupCluster() throws Exception {
-        safeGet(
-            clusterAdmin().execute(
-                DeletePipelineTransportAction.TYPE,
-                new DeletePipelineRequest(
-                    TEST_REQUEST_TIMEOUT,
-                    TEST_REQUEST_TIMEOUT,
-                    MigrateTemplateRegistry.REINDEX_DATA_STREAM_PIPELINE_NAME
-                )
-            )
-        );
-        super.cleanUpCluster();
+    private void cleanup() {
+        deletePipeline(MigrateTemplateRegistry.REINDEX_DATA_STREAM_PIPELINE_NAME);
     }
 
     private static final String MAPPING = """

Original file line number	Diff line number	Diff line change
`@@ -75,7 +75,7 @@ protected Collection<String> allClusters() {`
`75`	`75`	`@Override`
`76`	`76`	`protected Collection<Class<? extends Plugin>> nodePlugins(String clusterAlias) {`
`77`	`77`	`List<Class<? extends Plugin>> plugins = new ArrayList<>(super.nodePlugins(clusterAlias));`
`78`		`- plugins.add(CrossClustersEnrichIT.LocalStateEnrich.class);`
	`78`	`+ plugins.add(CrossClusterEnrichIT.LocalStateEnrich.class);`
`79`	`79`	`plugins.add(IngestCommonPlugin.class);`
`80`	`80`	`plugins.add(ReindexPlugin.class);`
`81`	`81`	`return plugins;`
`@@ -274,7 +274,7 @@ protected List<ActionType<XPackInfoFeatureResponse>> infoActions() {`
`274`	`274`
`275`	`275`	`@Override`
`276`	`276`	`protected Class<? extends TransportAction<XPackInfoRequest, XPackInfoResponse>> getInfoAction() {`
`277`		`- return CrossClustersQueriesWithInvalidLicenseIT.LocalStateEnrich.EnrichTransportXPackInfoAction.class;`
	`277`	`+ return CrossClusterQueriesWithInvalidLicenseIT.LocalStateEnrich.EnrichTransportXPackInfoAction.class;`
`278`	`278`	`}`
`279`	`279`	`}`
`280`	`280`	`}`