Merge branch 'main' into arbitrary-intervals-of-month-year-buckets

Author: kanoshiou

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: []

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,
     }
 

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

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

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

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -4491,14 +4491,17 @@ public void waitForEngineOrClosedShard(ActionListener<Void> listener) {
     }
 
     /**
-     * Registers a listener for an event when the shard advances to the provided primary term and segment generation
+     * Registers a listener for an event when the shard advances to the provided primary term and segment generation.
+     * Completes the listener with a {@link IndexShardClosedException} if the shard is closed.
      */
     public void waitForPrimaryTermAndGeneration(long primaryTerm, long segmentGeneration, ActionListener<Long> listener) {
-        waitForEngineOrClosedShard(
-            listener.delegateFailureAndWrap(
-                (l, ignored) -> getEngine().addPrimaryTermAndGenerationListener(primaryTerm, segmentGeneration, l)
-            )
-        );
+        waitForEngineOrClosedShard(listener.delegateFailureAndWrap((l, ignored) -> {
+            if (state == IndexShardState.CLOSED) {
+                l.onFailure(new IndexShardClosedException(shardId));
+            } else {
+                getEngine().addPrimaryTermAndGenerationListener(primaryTerm, segmentGeneration, l);
+            }
+        }));
     }
 
     /**

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

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

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

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