Merge remote-tracking branch 'elastic/main' into batched-exec-short

original-brownbear · original-brownbear · commit e409dae6fb89 · 2025-02-27T17:54:38.000+01:00
diff --git a/build-tools-internal/src/main/groovy/elasticsearch.build-scan.gradle b/build-tools-internal/src/main/groovy/elasticsearch.build-scan.gradle
@@ -133,6 +133,9 @@ develocity {
         }
       } else {
         tag 'LOCAL'
+        if (providers.systemProperty('idea.active').present) {
+          tag 'IDEA'
+        }
       }
     }
   }
diff --git a/build-tools-internal/src/main/groovy/elasticsearch.ide.gradle b/build-tools-internal/src/main/groovy/elasticsearch.ide.gradle
@@ -179,7 +179,7 @@ if (providers.systemProperty('idea.active').getOrNull() == 'true') {
 
   // this path is produced by the extractLibs task above
   String testLibraryPath = TestUtil.getTestLibraryPath("${elasticsearchProject.left()}/libs/native/libraries/build/platform")
-
+  def enableIdeaCC = providers.gradleProperty("org.elasticsearch.idea-configuration-cache").getOrElse("true").toBoolean()
   idea {
     project {
       vcs = 'Git'
@@ -209,6 +209,11 @@ if (providers.systemProperty('idea.active').getOrNull() == 'true') {
           }
         }
         runConfigurations {
+          defaults(org.jetbrains.gradle.ext.Gradle) {
+            scriptParameters = enableIdeaCC ? [
+              '--configuration-cache'
+            ].join(' ') : ''
+          }
           defaults(JUnit) {
             vmParameters = [
               '-ea',
diff --git a/docs/changelog/123569.yaml b/docs/changelog/123569.yaml
@@ -0,0 +1,5 @@
+pr: 123569
+summary: Abort pending deletion on `IndicesService` close
+area: Store
+type: enhancement
+issues: []
diff --git a/gradle.properties b/gradle.properties
@@ -19,3 +19,6 @@ org.gradle.dependency.verification.console=verbose
 
 # allow user to specify toolchain via the RUNTIME_JAVA_HOME environment variable
 org.gradle.java.installations.fromEnv=RUNTIME_JAVA_HOME
+
+# if configuration cache enabled then enable parallel support too
+org.gradle.configuration-cache.parallel=true
diff --git a/libs/ssl-config/build.gradle b/libs/ssl-config/build.gradle
@@ -10,6 +10,7 @@ apply plugin: "elasticsearch.publish"
 
 dependencies {
   api project(':libs:core')
+  api project(':libs:entitlement')
 
   testImplementation(project(":test:framework")) {
     exclude group: 'org.elasticsearch', module: 'ssl-config'
diff --git a/libs/ssl-config/src/main/java/module-info.java b/libs/ssl-config/src/main/java/module-info.java
@@ -9,6 +9,7 @@
 
 module org.elasticsearch.sslconfig {
     requires org.elasticsearch.base;
+    requires org.elasticsearch.entitlement;
 
     exports org.elasticsearch.common.ssl;
 }
diff --git a/libs/ssl-config/src/main/java/org/elasticsearch/common/ssl/PemKeyConfig.java b/libs/ssl-config/src/main/java/org/elasticsearch/common/ssl/PemKeyConfig.java
@@ -10,6 +10,7 @@
 package org.elasticsearch.common.ssl;
 
 import org.elasticsearch.core.Tuple;
+import org.elasticsearch.entitlement.runtime.api.NotEntitledException;
 
 import java.io.IOException;
 import java.nio.file.Path;
@@ -127,6 +128,8 @@ private PrivateKey getPrivateKey(Path path) {
             return privateKey;
         } catch (AccessControlException e) {
             throw SslFileUtil.accessControlFailure(KEY_FILE_TYPE, List.of(path), e, configBasePath);
+        } catch (NotEntitledException e) {
+            throw SslFileUtil.notEntitledFailure(KEY_FILE_TYPE, List.of(path), e, configBasePath);
         } catch (IOException e) {
             throw SslFileUtil.ioException(KEY_FILE_TYPE, List.of(path), e);
         } catch (GeneralSecurityException e) {
diff --git a/libs/ssl-config/src/main/java/org/elasticsearch/common/ssl/PemTrustConfig.java b/libs/ssl-config/src/main/java/org/elasticsearch/common/ssl/PemTrustConfig.java
@@ -9,6 +9,8 @@
 
 package org.elasticsearch.common.ssl;
 
+import org.elasticsearch.entitlement.runtime.api.NotEntitledException;
+
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.file.Path;
@@ -99,6 +101,8 @@ private List<Certificate> readCertificates(List<Path> paths) {
             return PemUtils.readCertificates(paths);
         } catch (AccessControlException e) {
             throw SslFileUtil.accessControlFailure(CA_FILE_TYPE, paths, e, basePath);
+        } catch (NotEntitledException e) {
+            throw SslFileUtil.notEntitledFailure(CA_FILE_TYPE, paths, e, basePath);
         } catch (IOException e) {
             throw SslFileUtil.ioException(CA_FILE_TYPE, paths, e);
         } catch (GeneralSecurityException e) {
diff --git a/libs/ssl-config/src/main/java/org/elasticsearch/common/ssl/PemUtils.java b/libs/ssl-config/src/main/java/org/elasticsearch/common/ssl/PemUtils.java
@@ -10,6 +10,7 @@
 package org.elasticsearch.common.ssl;
 
 import org.elasticsearch.core.CharArrays;
+import org.elasticsearch.entitlement.runtime.api.NotEntitledException;
 
 import java.io.BufferedReader;
 import java.io.IOException;
@@ -112,6 +113,8 @@ public static PrivateKey readPrivateKey(Path path, Supplier<char[]> passwordSupp
             return privateKey;
         } catch (AccessControlException e) {
             throw SslFileUtil.accessControlFailure("PEM private key", List.of(path), e, null);
+        } catch (NotEntitledException e) {
+            throw SslFileUtil.notEntitledFailure("PEM private key", List.of(path), e, null);
         } catch (IOException e) {
             throw SslFileUtil.ioException("PEM private key", List.of(path), e);
         } catch (GeneralSecurityException e) {
diff --git a/libs/ssl-config/src/main/java/org/elasticsearch/common/ssl/SslFileUtil.java b/libs/ssl-config/src/main/java/org/elasticsearch/common/ssl/SslFileUtil.java
@@ -9,6 +9,8 @@
 
 package org.elasticsearch.common.ssl;
 
+import org.elasticsearch.entitlement.runtime.api.NotEntitledException;
+
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.nio.file.AccessDeniedException;
@@ -78,7 +80,15 @@ static SslConfigException accessDenied(String fileType, List<Path> paths, Access
         return new SslConfigException(message, cause);
     }
 
+    static SslConfigException notEntitledFailure(String fileType, List<Path> paths, NotEntitledException cause, Path basePath) {
+        return innerAccessControlFailure(fileType, paths, cause, basePath);
+    }
+
     static SslConfigException accessControlFailure(String fileType, List<Path> paths, AccessControlException cause, Path basePath) {
+        return innerAccessControlFailure(fileType, paths, cause, basePath);
+    }
+
+    private static SslConfigException innerAccessControlFailure(String fileType, List<Path> paths, Exception cause, Path basePath) {
         String message = "cannot read configured " + fileType + " [" + pathsToString(paths) + "] because ";
         if (paths.size() == 1) {
             message += "access to read the file is blocked";
diff --git a/libs/ssl-config/src/main/java/org/elasticsearch/common/ssl/StoreKeyConfig.java b/libs/ssl-config/src/main/java/org/elasticsearch/common/ssl/StoreKeyConfig.java
@@ -11,6 +11,7 @@
 
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Tuple;
+import org.elasticsearch.entitlement.runtime.api.NotEntitledException;
 
 import java.io.IOException;
 import java.nio.file.Path;
@@ -168,6 +169,8 @@ private KeyStore readKeyStore(Path path) {
             return KeyStoreUtil.readKeyStore(path, type, storePassword);
         } catch (AccessControlException e) {
             throw SslFileUtil.accessControlFailure("[" + type + "] keystore", List.of(path), e, configBasePath);
+        } catch (NotEntitledException e) {
+            throw SslFileUtil.notEntitledFailure("[" + type + "] keystore", List.of(path), e, configBasePath);
         } catch (IOException e) {
             throw SslFileUtil.ioException("[" + type + "] keystore", List.of(path), e);
         } catch (GeneralSecurityException e) {
diff --git a/libs/ssl-config/src/main/java/org/elasticsearch/common/ssl/StoreTrustConfig.java b/libs/ssl-config/src/main/java/org/elasticsearch/common/ssl/StoreTrustConfig.java
@@ -9,6 +9,8 @@
 
 package org.elasticsearch.common.ssl;
 
+import org.elasticsearch.entitlement.runtime.api.NotEntitledException;
+
 import java.io.IOException;
 import java.nio.file.Path;
 import java.security.AccessControlException;
@@ -95,6 +97,8 @@ private KeyStore readKeyStore(Path path) {
             return KeyStoreUtil.readKeyStore(path, type, password);
         } catch (AccessControlException e) {
             throw SslFileUtil.accessControlFailure(fileTypeForException(), List.of(path), e, configBasePath);
+        } catch (NotEntitledException e) {
+            throw SslFileUtil.notEntitledFailure(fileTypeForException(), List.of(path), e, configBasePath);
         } catch (IOException e) {
             throw SslFileUtil.ioException(fileTypeForException(), List.of(path), e, getAdditionalErrorDetails());
         } catch (GeneralSecurityException e) {
diff --git a/muted-tests.yml b/muted-tests.yml
@@ -254,8 +254,6 @@ tests:
 - class: org.elasticsearch.test.rest.ClientYamlTestSuiteIT
   method: test {yaml=snapshot.delete/10_basic/Delete a snapshot asynchronously}
   issue: https://github.com/elastic/elasticsearch/issues/122102
-- 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
@@ -286,15 +284,9 @@ tests:
 - class: org.elasticsearch.smoketest.DocsClientYamlTestSuiteIT
   method: test {yaml=reference/snapshot-restore/apis/get-snapshot-api/line_408}
   issue: https://github.com/elastic/elasticsearch/issues/122681
-- class: org.elasticsearch.xpack.autoscaling.storage.ReactiveStorageIT
-  method: testScaleWhileShrinking
-  issue: https://github.com/elastic/elasticsearch/issues/122119
 - class: org.elasticsearch.search.basic.SearchWithRandomDisconnectsIT
   method: testSearchWithRandomDisconnects
   issue: https://github.com/elastic/elasticsearch/issues/122707
-- class: org.elasticsearch.snapshots.DedicatedClusterSnapshotRestoreIT
-  method: testRestoreShrinkIndex
-  issue: https://github.com/elastic/elasticsearch/issues/121717
 - class: org.elasticsearch.smoketest.DocsClientYamlTestSuiteIT
   method: test {yaml=reference/cat/allocation/cat-allocation-example}
   issue: https://github.com/elastic/elasticsearch/issues/121976
@@ -340,6 +332,9 @@ tests:
 - class: org.elasticsearch.xpack.esql.action.CrossClusterQueryWithPartialResultsIT
   method: testOneRemoteClusterPartial
   issue: https://github.com/elastic/elasticsearch/issues/123451
+- class: org.elasticsearch.action.admin.cluster.node.tasks.CancellableTasksIT
+  method: testChildrenTasksCancelledOnTimeout
+  issue: https://github.com/elastic/elasticsearch/issues/123568
 
 # Examples:
 #
diff --git a/server/src/main/java/org/elasticsearch/indices/IndicesService.java b/server/src/main/java/org/elasticsearch/indices/IndicesService.java
@@ -255,6 +255,7 @@ public class IndicesService extends AbstractLifecycleComponent
     private final Map<String, IndexStorePlugin.RecoveryStateFactory> recoveryStateFactories;
     private final IndexStorePlugin.IndexFoldersDeletionListener indexFoldersDeletionListeners;
     final AbstractRefCounted indicesRefCount; // pkg-private for testing
+    private final CountDownLatch stopLatch = new CountDownLatch(1);
     private final CountDownLatch closeLatch = new CountDownLatch(1);
     private volatile boolean idFieldDataEnabled;
     private volatile boolean allowExpensiveQueries;
@@ -403,6 +404,7 @@ public ClusterService clusterService() {
 
     @Override
     protected void doStop() {
+        stopLatch.countDown();
         clusterService.removeApplier(timestampFieldMapperService);
         timestampFieldMapperService.doStop();
 
@@ -1440,7 +1442,15 @@ public void processPendingDeletes(Index index, IndexSettings indexSettings, Time
                     }
                     if (remove.isEmpty() == false) {
                         logger.warn("{} still pending deletes present for shards {} - retrying", index, remove.toString());
-                        Thread.sleep(sleepTime);
+                        if (stopLatch.await(sleepTime, TimeUnit.MILLISECONDS)) {
+                            logger.info(
+                                "Indices service stopped. {} aborting pending deletes after [{}] for shards {}",
+                                index,
+                                TimeValue.timeValueNanos(System.nanoTime() - startTimeNS),
+                                remove.toString()
+                            );
+                            break;
+                        }
                         sleepTime = Math.min(maxSleepTimeMs, sleepTime * 2); // increase the sleep time gradually
                         logger.debug("{} schedule pending delete retry after {} ms", index, sleepTime);
                     }
diff --git a/server/src/main/java/org/elasticsearch/search/internal/ContextIndexSearcher.java b/server/src/main/java/org/elasticsearch/search/internal/ContextIndexSearcher.java
@@ -36,7 +36,6 @@
 import org.apache.lucene.util.BitSetIterator;
 import org.apache.lucene.util.Bits;
 import org.apache.lucene.util.SparseFixedBitSet;
-import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.lucene.util.CombinedBitSet;
 import org.elasticsearch.search.dfs.AggregatedDfs;
@@ -53,7 +52,6 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.PriorityQueue;
-import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.Executor;
 import java.util.stream.Collectors;
@@ -82,7 +80,6 @@ public class ContextIndexSearcher extends IndexSearcher implements Releasable {
     // don't create slices with less than this number of docs
     private final int minimumDocsPerSlice;
 
-    private final Set<Thread> timeoutOverwrites = ConcurrentCollections.newConcurrentSet();
     private volatile boolean timeExceeded = false;
 
     /** constructor for non-concurrent search */
@@ -374,6 +371,8 @@ private <C extends Collector, T> T search(Weight weight, CollectorManager<C, T>
         }
     }
 
+    private static final ThreadLocal<Boolean> timeoutOverwrites = ThreadLocal.withInitial(() -> false);
+
     /**
      * Similar to the lucene implementation, with the following changes made:
      * 1) postCollection is performed after each segment is collected. This is needed for aggregations, performed by search threads
@@ -397,12 +396,12 @@ public void search(LeafReaderContextPartition[] leaves, Weight weight, Collector
                 try {
                     // Search phase has finished, no longer need to check for timeout
                     // otherwise the aggregation post-collection phase might get cancelled.
-                    boolean added = timeoutOverwrites.add(Thread.currentThread());
-                    assert added;
+                    assert timeoutOverwrites.get() == false;
+                    timeoutOverwrites.set(true);
                     doAggregationPostCollection(collector);
                 } finally {
-                    boolean removed = timeoutOverwrites.remove(Thread.currentThread());
-                    assert removed;
+                    assert timeoutOverwrites.get();
+                    timeoutOverwrites.set(false);
                 }
             }
         }
@@ -420,7 +419,7 @@ public boolean timeExceeded() {
     }
 
     public void throwTimeExceededException() {
-        if (timeoutOverwrites.contains(Thread.currentThread()) == false) {
+        if (timeoutOverwrites.get() == false) {
             throw new TimeExceededException();
         }
     }
diff --git a/x-pack/plugin/core/src/main/java/module-info.java b/x-pack/plugin/core/src/main/java/module-info.java
@@ -7,6 +7,7 @@
 
 module org.elasticsearch.xcore {
     requires org.elasticsearch.cli;
+    requires org.elasticsearch.entitlement;
     requires org.elasticsearch.base;
     requires org.elasticsearch.grok;
     requires org.elasticsearch.server;
diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ssl/SSLConfigurationReloader.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ssl/SSLConfigurationReloader.java
@@ -12,6 +12,7 @@
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.common.ssl.SslConfiguration;
 import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.entitlement.runtime.api.NotEntitledException;
 import org.elasticsearch.watcher.FileChangesListener;
 import org.elasticsearch.watcher.FileWatcher;
 import org.elasticsearch.watcher.ResourceWatcherService;
@@ -109,7 +110,7 @@ private static void startWatching(
             fileWatcher.addListener(changeListener);
             try {
                 resourceWatcherService.add(fileWatcher, Frequency.HIGH);
-            } catch (IOException | AccessControlException e) {
+            } catch (IOException | AccessControlException | NotEntitledException e) {
                 logger.error("failed to start watching directory [{}] for ssl configurations [{}] - {}", path, configurations, e);
             }
         });
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/LocalExecutionPlanner.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/LocalExecutionPlanner.java
@@ -198,8 +198,8 @@ public LocalExecutionPlan plan(String taskDescription, FoldContext foldCtx, Phys
             new DriverFactory(
                 new DriverSupplier(
                     taskDescription,
-                    Node.NODE_NAME_SETTING.get(settings),
                     ClusterName.CLUSTER_NAME_SETTING.get(settings).value(),
+                    Node.NODE_NAME_SETTING.get(settings),
                     context.bigArrays,
                     context.blockFactory,
                     physicalOperation,
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/planner/LocalExecutionPlannerTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/planner/LocalExecutionPlannerTests.java
@@ -17,6 +17,7 @@
 import org.apache.lucene.search.IndexSearcher;
 import org.apache.lucene.store.Directory;
 import org.apache.lucene.tests.index.RandomIndexWriter;
+import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.compute.lucene.LuceneSourceOperator;
@@ -29,6 +30,7 @@
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.index.cache.query.TrivialQueryCachingPolicy;
 import org.elasticsearch.index.mapper.MapperServiceTestCase;
+import org.elasticsearch.node.Node;
 import org.elasticsearch.search.internal.AliasFilter;
 import org.elasticsearch.search.internal.ContextIndexSearcher;
 import org.elasticsearch.xpack.esql.core.expression.FieldAttribute;
@@ -159,6 +161,29 @@ public void testLuceneTopNSourceOperatorDistanceSort() throws IOException {
         assertThat(factory.limit(), equalTo(10));
     }
 
+    public void testDriverClusterAndNodeName() throws IOException {
+        int estimatedRowSize = randomEstimatedRowSize(estimatedRowSizeIsHuge);
+        LocalExecutionPlanner.LocalExecutionPlan plan = planner().plan(
+            "test",
+            FoldContext.small(),
+            new EsQueryExec(
+                Source.EMPTY,
+                index().name(),
+                IndexMode.STANDARD,
+                index().indexNameWithModes(),
+                List.of(),
+                null,
+                null,
+                null,
+                estimatedRowSize
+            )
+        );
+        assertThat(plan.driverFactories.size(), lessThanOrEqualTo(pragmas.taskConcurrency()));
+        LocalExecutionPlanner.DriverSupplier supplier = plan.driverFactories.get(0).driverSupplier();
+        assertThat(supplier.clusterName(), equalTo("dev-cluster"));
+        assertThat(supplier.nodeName(), equalTo("node-1"));
+    }
+
     private int randomEstimatedRowSize(boolean huge) {
         int hugeBoundary = SourceOperator.MIN_TARGET_PAGE_SIZE * 10;
         return huge ? between(hugeBoundary, Integer.MAX_VALUE) : between(1, hugeBoundary);
@@ -179,7 +204,10 @@ private LocalExecutionPlanner planner() throws IOException {
             null,
             BigArrays.NON_RECYCLING_INSTANCE,
             TestBlockFactory.getNonBreakingInstance(),
-            Settings.EMPTY,
+            Settings.builder()
+                .put(ClusterName.CLUSTER_NAME_SETTING.getKey(), "dev-cluster")
+                .put(Node.NODE_NAME_SETTING.getKey(), "node-1")
+                .build(),
             config(),
             null,
             null,
diff --git a/x-pack/plugin/ilm/qa/multi-node/src/javaRestTest/java/org/elasticsearch/xpack/TimeSeriesRestDriver.java b/x-pack/plugin/ilm/qa/multi-node/src/javaRestTest/java/org/elasticsearch/xpack/TimeSeriesRestDriver.java
diff --git a/x-pack/plugin/ilm/qa/multi-node/src/javaRestTest/java/org/elasticsearch/xpack/ilm/TimeSeriesDataStreamsIT.java b/x-pack/plugin/ilm/qa/multi-node/src/javaRestTest/java/org/elasticsearch/xpack/ilm/TimeSeriesDataStreamsIT.java
diff --git a/x-pack/plugin/security/qa/security-trial/src/javaRestTest/java/org/elasticsearch/xpack/security/ssl/SslEntitlementRestIT.java b/x-pack/plugin/security/qa/security-trial/src/javaRestTest/java/org/elasticsearch/xpack/security/ssl/SslEntitlementRestIT.java

Original file line number	Diff line number	Diff line change
`@@ -133,6 +133,9 @@ develocity {`
`133`	`133`	`}`
`134`	`134`	`} else {`
`135`	`135`	`tag 'LOCAL'`
	`136`	`+ if (providers.systemProperty('idea.active').present) {`
	`137`	`+ tag 'IDEA'`
	`138`	`+ }`
`136`	`139`	`}`
`137`	`140`	`}`
`138`	`141`	`}`
Original file line number	Diff line number	Diff line change
`@@ -9,6 +9,7 @@`
`9`	`9`
`10`	`10`	`module org.elasticsearch.sslconfig {`
`11`	`11`	`requires org.elasticsearch.base;`
	`12`	`+ requires org.elasticsearch.entitlement;`
`12`	`13`
`13`	`14`	`exports org.elasticsearch.common.ssl;`
`14`	`15`	`}`