Merge branch 'main' into read-failure-store-privilege-role-building

n1v0lg · web-flow · commit d470e5af675c · 2025-02-19T13:41:04.000+01:00
diff --git a/docs/changelog/122731.yaml b/docs/changelog/122731.yaml
@@ -0,0 +1,5 @@
+pr: 122731
+summary: Fork post-snapshot-delete cleanup off master thread
+area: Snapshot/Restore
+type: bug
+issues: []
diff --git a/libs/core/src/main/java/org/elasticsearch/core/UpdateForV9.java b/libs/core/src/main/java/org/elasticsearch/core/UpdateForV9.java
@@ -24,14 +24,11 @@
     enum Owner {
         CORE_INFRA,
         DATA_MANAGEMENT,
-        DISTRIBUTED_COORDINATION,
         DISTRIBUTED_INDEXING,
         ENTERPRISE_SEARCH,
         MACHINE_LEARNING,
         PROFILING,
         SEARCH_ANALYTICS,
-        SEARCH_FOUNDATIONS,
-        SEARCH_RELEVANCE,
         SECURITY,
     }
 
diff --git a/muted-tests.yml b/muted-tests.yml
@@ -61,9 +61,6 @@ tests:
 - class: org.elasticsearch.xpack.test.rest.XPackRestIT
   method: test {p0=transform/transforms_start_stop/Verify start transform reuses destination index}
   issue: https://github.com/elastic/elasticsearch/issues/115808
-- class: org.elasticsearch.search.StressSearchServiceReaperIT
-  method: testStressReaper
-  issue: https://github.com/elastic/elasticsearch/issues/115816
 - class: org.elasticsearch.xpack.application.connector.ConnectorIndexServiceTests
   issue: https://github.com/elastic/elasticsearch/issues/116087
 - class: org.elasticsearch.xpack.test.rest.XPackRestIT
diff --git a/rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/snapshot.delete/10_basic.yml b/rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/snapshot.delete/10_basic.yml
@@ -68,3 +68,10 @@ setup:
         wait_for_completion: false
 
   - match: { acknowledged: true }
+
+  # now create another snapshot just to ensure that the async delete finishes before the test cleanup runs:
+  - do:
+      snapshot.create:
+        repository: test_repo_create_1
+        snapshot: barrier_snapshot
+        wait_for_completion: true
diff --git a/server/src/internalClusterTest/java/org/elasticsearch/snapshots/RepositoriesIT.java b/server/src/internalClusterTest/java/org/elasticsearch/snapshots/RepositoriesIT.java
@@ -508,6 +508,12 @@ public void taskSucceeded(ClusterStateTaskListener clusterStateTaskListener, Obj
             .orElseThrow()
             .queue();
 
+        // There is one task in the queue for computing and forking the cleanup work.
+        assertThat(queueLength.getAsInt(), equalTo(1));
+
+        safeAwait(barrier); // unblock the barrier thread and let it process the queue
+        safeAwait(barrier); // wait for the queue to be processed
+
         // There are indexCount (=3*snapshotPoolSize) index-deletion tasks, plus one for cleaning up the root metadata. However, the
         // throttled runner only enqueues one task per SNAPSHOT thread to start with, and then the eager runner adds another one. This shows
         // we are not spamming the threadpool with all the tasks at once, which means that other snapshot activities can run alongside this
diff --git a/server/src/main/java/org/elasticsearch/repositories/RepositoryData.java b/server/src/main/java/org/elasticsearch/repositories/RepositoryData.java
@@ -29,6 +29,7 @@
 import org.elasticsearch.snapshots.SnapshotInfo;
 import org.elasticsearch.snapshots.SnapshotState;
 import org.elasticsearch.snapshots.SnapshotsService;
+import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
 
@@ -377,6 +378,7 @@ private static boolean isIndexToUpdateAfterRemovingSnapshots(
      * @return map of index to index metadata blob id to delete
      */
     public Map<IndexId, Collection<String>> indexMetaDataToRemoveAfterRemovingSnapshots(Collection<SnapshotId> snapshotIds) {
+        assert ThreadPool.assertCurrentThreadPool(ThreadPool.Names.SNAPSHOT);
         Iterator<IndexId> indicesForSnapshot = indicesToUpdateAfterRemovingSnapshot(snapshotIds);
         final Set<String> allRemainingIdentifiers = indexMetaDataGenerations.lookup.entrySet()
             .stream()
diff --git a/server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java b/server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java
@@ -1134,14 +1134,20 @@ private void runWithUniqueShardMetadataNaming(ActionListener<RepositoryData> rep
                     );
                 })
 
-                .<RepositoryData>andThen((l, newRepositoryData) -> {
-                    l.onResponse(newRepositoryData);
-                    // Once we have updated the repository, run the unreferenced blobs cleanup in parallel to shard-level snapshot deletion
-                    try (var refs = new RefCountingRunnable(onCompletion)) {
-                        cleanupUnlinkedRootAndIndicesBlobs(newRepositoryData, refs.acquireListener());
-                        cleanupUnlinkedShardLevelBlobs(refs.acquireListener());
+                .<RepositoryData>andThen(
+                    // writeIndexGen finishes on master-service thread so must fork here.
+                    snapshotExecutor,
+                    threadPool.getThreadContext(),
+                    (l, newRepositoryData) -> {
+                        l.onResponse(newRepositoryData);
+                        // Once we have updated the repository, run the unreferenced blobs cleanup in parallel to shard-level snapshot
+                        // deletion
+                        try (var refs = new RefCountingRunnable(onCompletion)) {
+                            cleanupUnlinkedRootAndIndicesBlobs(newRepositoryData, refs.acquireListener());
+                            cleanupUnlinkedShardLevelBlobs(refs.acquireListener());
+                        }
                     }
-                })
+                )
 
                 .addListener(repositoryDataUpdateListener);
         }
diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/metrics/AvgAggregator.java b/server/src/main/java/org/elasticsearch/search/aggregations/metrics/AvgAggregator.java
@@ -58,7 +58,7 @@ protected LeafBucketCollector getLeafCollector(NumericDoubleValues values, final
             public void collect(int doc, long bucket) throws IOException {
                 if (values.advanceExact(doc)) {
                     maybeGrow(bucket);
-                    computeSum(bucket, values, sums, compensations);
+                    computeSum(bucket, values.doubleValue(), sums, compensations);
                     counts.increment(bucket, 1L);
                 }
             }
diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/metrics/ExtendedStatsAggregator.java b/server/src/main/java/org/elasticsearch/search/aggregations/metrics/ExtendedStatsAggregator.java
@@ -105,7 +105,6 @@ public void collect(int doc, long bucket) throws IOException {
 
     @Override
     protected LeafBucketCollector getLeafCollector(NumericDoubleValues values, final LeafBucketCollector sub) {
-        final CompensatedSum compensatedSum = new CompensatedSum(0, 0);
         return new LeafBucketCollectorBase(sub, values) {
 
             @Override
@@ -114,20 +113,9 @@ public void collect(int doc, long bucket) throws IOException {
                     maybeGrow(bucket);
                     final double value = values.doubleValue();
                     counts.increment(bucket, 1L);
-                    // Compute the sum and sum of squires for double values with Kahan summation algorithm
-                    // which is more accurate than naive summation.
-                    compensatedSum.reset(sums.get(bucket), compensations.get(bucket));
-                    compensatedSum.add(value);
-                    sums.set(bucket, compensatedSum.value());
-                    compensations.set(bucket, compensatedSum.delta());
-
-                    compensatedSum.reset(sumOfSqrs.get(bucket), compensationOfSqrs.get(bucket));
-                    compensatedSum.add(value * value);
-                    sumOfSqrs.set(bucket, compensatedSum.value());
-                    compensationOfSqrs.set(bucket, compensatedSum.delta());
-
-                    mins.set(bucket, Math.min(mins.get(bucket), value));
-                    maxes.set(bucket, Math.max(maxes.get(bucket), value));
+                    SumAggregator.computeSum(bucket, value, sums, compensations);
+                    SumAggregator.computeSum(bucket, value * value, sumOfSqrs, compensationOfSqrs);
+                    StatsAggregator.updateMinsAndMaxes(bucket, value, mins, maxes);
                 }
             }
 
@@ -138,13 +126,14 @@ private void maybeGrow(long bucket) {
         if (bucket >= counts.size()) {
             final long from = counts.size();
             final long overSize = BigArrays.overSize(bucket + 1);
-            counts = bigArrays().resize(counts, overSize);
-            sums = bigArrays().resize(sums, overSize);
-            compensations = bigArrays().resize(compensations, overSize);
-            mins = bigArrays().resize(mins, overSize);
-            maxes = bigArrays().resize(maxes, overSize);
-            sumOfSqrs = bigArrays().resize(sumOfSqrs, overSize);
-            compensationOfSqrs = bigArrays().resize(compensationOfSqrs, overSize);
+            var bigArrays = bigArrays();
+            counts = bigArrays.resize(counts, overSize);
+            sums = bigArrays.resize(sums, overSize);
+            compensations = bigArrays.resize(compensations, overSize);
+            mins = bigArrays.resize(mins, overSize);
+            maxes = bigArrays.resize(maxes, overSize);
+            sumOfSqrs = bigArrays.resize(sumOfSqrs, overSize);
+            compensationOfSqrs = bigArrays.resize(compensationOfSqrs, overSize);
             mins.fill(from, overSize, Double.POSITIVE_INFINITY);
             maxes.fill(from, overSize, Double.NEGATIVE_INFINITY);
         }
diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/metrics/GeoCentroidAggregator.java b/server/src/main/java/org/elasticsearch/search/aggregations/metrics/GeoCentroidAggregator.java
@@ -93,7 +93,6 @@ public void collect(int doc, long bucket) throws IOException {
     }
 
     private LeafBucketCollector getLeafCollector(GeoPointValues values, LeafBucketCollector sub) {
-        final CompensatedSum compensatedSum = new CompensatedSum(0, 0);
         return new LeafBucketCollectorBase(sub, values) {
             @Override
             public void collect(int doc, long bucket) throws IOException {
@@ -104,16 +103,8 @@ public void collect(int doc, long bucket) throws IOException {
                     // Compute the sum of double values with Kahan summation algorithm which is more
                     // accurate than naive summation.
                     final GeoPoint value = values.pointValue();
-                    // latitude
-                    compensatedSum.reset(latSum.get(bucket), latCompensations.get(bucket));
-                    compensatedSum.add(value.getLat());
-                    latSum.set(bucket, compensatedSum.value());
-                    latCompensations.set(bucket, compensatedSum.delta());
-                    // longitude
-                    compensatedSum.reset(lonSum.get(bucket), lonCompensations.get(bucket));
-                    compensatedSum.add(value.getLon());
-                    lonSum.set(bucket, compensatedSum.value());
-                    lonCompensations.set(bucket, compensatedSum.delta());
+                    SumAggregator.computeSum(bucket, value.getLat(), latSum, latCompensations);
+                    SumAggregator.computeSum(bucket, value.getLon(), lonSum, lonCompensations);
                 }
             }
         };
diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/metrics/StatsAggregator.java b/server/src/main/java/org/elasticsearch/search/aggregations/metrics/StatsAggregator.java
@@ -81,36 +81,43 @@ public void collect(int doc, long bucket) throws IOException {
 
     @Override
     public LeafBucketCollector getLeafCollector(NumericDoubleValues values, LeafBucketCollector sub) {
-        final CompensatedSum kahanSummation = new CompensatedSum(0, 0);
         return new LeafBucketCollectorBase(sub, values) {
             @Override
             public void collect(int doc, long bucket) throws IOException {
                 if (values.advanceExact(doc)) {
                     maybeGrow(bucket);
                     counts.increment(bucket, 1L);
-                    // Compute the sum of double values with Kahan summation algorithm which is more
-                    // accurate than naive summation.
-                    kahanSummation.reset(sums.get(bucket), compensations.get(bucket));
                     double value = values.doubleValue();
-                    kahanSummation.add(value);
-                    sums.set(bucket, kahanSummation.value());
-                    compensations.set(bucket, kahanSummation.delta());
-                    mins.set(bucket, Math.min(mins.get(bucket), value));
-                    maxes.set(bucket, Math.max(maxes.get(bucket), value));
+                    SumAggregator.computeSum(bucket, value, sums, compensations);
+                    updateMinsAndMaxes(bucket, value, mins, maxes);
                 }
             }
         };
     }
 
+    static void updateMinsAndMaxes(long bucket, double value, DoubleArray mins, DoubleArray maxes) {
+        double min = mins.get(bucket);
+        double updated = Math.min(value, min);
+        if (updated != min) {
+            mins.set(bucket, updated);
+        }
+        double max = maxes.get(bucket);
+        updated = Math.max(value, max);
+        if (updated != max) {
+            maxes.set(bucket, updated);
+        }
+    }
+
     private void maybeGrow(long bucket) {
         if (bucket >= counts.size()) {
             final long from = counts.size();
             final long overSize = BigArrays.overSize(bucket + 1);
-            counts = bigArrays().resize(counts, overSize);
-            sums = bigArrays().resize(sums, overSize);
-            compensations = bigArrays().resize(compensations, overSize);
-            mins = bigArrays().resize(mins, overSize);
-            maxes = bigArrays().resize(maxes, overSize);
+            var bigArrays = bigArrays();
+            counts = bigArrays.resize(counts, overSize);
+            sums = bigArrays.resize(sums, overSize);
+            compensations = bigArrays.resize(compensations, overSize);
+            mins = bigArrays.resize(mins, overSize);
+            maxes = bigArrays.resize(maxes, overSize);
             mins.fill(from, overSize, Double.POSITIVE_INFINITY);
             maxes.fill(from, overSize, Double.NEGATIVE_INFINITY);
         }
diff --git a/server/src/main/java/org/elasticsearch/search/aggregations/metrics/SumAggregator.java b/server/src/main/java/org/elasticsearch/search/aggregations/metrics/SumAggregator.java
@@ -97,16 +97,15 @@ protected LeafBucketCollector getLeafCollector(NumericDoubleValues values, final
             public void collect(int doc, long bucket) throws IOException {
                 if (values.advanceExact(doc)) {
                     maybeGrow(bucket);
-                    computeSum(bucket, values, sums, compensations);
+                    computeSum(bucket, values.doubleValue(), sums, compensations);
                 }
             }
         };
     }
 
-    static void computeSum(long bucket, NumericDoubleValues values, DoubleArray sums, DoubleArray compensations) throws IOException {
+    static void computeSum(long bucket, double added, DoubleArray sums, DoubleArray compensations) {
         // Compute the sum of double values with Kahan summation algorithm which is more
         // accurate than naive summation.
-        double added = values.doubleValue();
         double value = addIfNonOrInf(added, sums.get(bucket));
         if (Double.isFinite(value)) {
             double delta = compensations.get(bucket);
@@ -122,8 +121,7 @@ static void computeSum(long bucket, NumericDoubleValues values, DoubleArray sums
 
     protected final void maybeGrow(long bucket) {
         if (bucket >= sums.size()) {
-            var bigArrays = bigArrays();
-            doGrow(bucket, bigArrays);
+            doGrow(bucket, bigArrays());
         }
     }
 

Original file line number	Diff line number	Diff line change
`@@ -58,7 +58,7 @@ protected LeafBucketCollector getLeafCollector(NumericDoubleValues values, final`
`58`	`58`	`public void collect(int doc, long bucket) throws IOException {`
`59`	`59`	`if (values.advanceExact(doc)) {`
`60`	`60`	`maybeGrow(bucket);`
`61`		`- computeSum(bucket, values, sums, compensations);`
	`61`	`+ computeSum(bucket, values.doubleValue(), sums, compensations);`
`62`	`62`	`counts.increment(bucket, 1L);`
`63`	`63`	`}`
`64`	`64`	`}`