Merge branch '8.19' into update-8.19-apm-data-resource-version

Author: ericywl

docs/changelog/130158.yaml

@@ -0,0 +1,5 @@
+pr: 130158
+summary: Handle unavailable MD5 in ES|QL
+area: ES|QL
+type: bug
+issues: []

docs/reference/index-modules/merge.asciidoc

@@ -14,18 +14,32 @@ resources between merging and other activities like search.
 [[merge-scheduling]]
 === Merge scheduling
 
-The merge scheduler (ConcurrentMergeScheduler) controls the execution of merge
-operations when they are needed. Merges run in separate threads, and when the
-maximum number of threads is reached, further merges will wait until a merge
-thread becomes available.
-
-The merge scheduler supports the following _dynamic_ setting:
-
-`index.merge.scheduler.max_thread_count`::
-
-    The maximum number of threads on a single shard that may be merging at once.
-	Defaults to
-    `Math.max(1, Math.min(4, <<node.processors, node.processors>> / 2))` which
-    works well for a good solid-state-disk (SSD). If your index is on spinning
-    platter drives instead, decrease this to 1.
+The merge scheduler controls the execution of merge operations when they are needed.
+Merges run on the dedicated `merge` thread pool.
+Smaller merges are prioritized over larger ones, across all shards on the node.
+Merges are disk IO throttled so that bursts, while merging activity is otherwise low, are smoothed out in order to not impact indexing throughput.
+There is no limit on the number of merges that can be enqueued for execution on the thread pool.
+However, beyond a certain per-shard limit, after merging is completely disk IO un-throttled, indexing for the shard will itself be throttled until merging catches up.
+
+The available disk space is periodically monitored, such that no new merge tasks are scheduled for execution when the available disk space is low.
+This is in order to prevent that the temporary disk space, which is required while merges are executed, completely fills up the disk space on the node.
+
+The merge scheduler supports the following *dynamic* settings:
+
+`index.merge.scheduler.max_thread_count`
+:   The maximum number of threads on a **single** shard that may be merging at once. Defaults to `Math.max(1, Math.min(4, <<node.processors, node.processors>> / 2))` which works well for a good solid-state-disk (SSD). If your index is on spinning platter drives instead, decrease this to 1.
+
+`indices.merge.disk.check_interval`
+:   The time interval for checking the available disk space. Defaults to `5s`.
+
+`indices.merge.disk.watermark.high`
+:   Controls the disk usage watermark, which defaults to `95%`, beyond which no merge tasks can start execution.
+The disk usage tally includes the estimated temporary disk space still required by all the currently executing merge tasks.
+Any merge task scheduled *before* the limit is reached continues execution, even if the limit is exceeded while executing
+(merge tasks are not aborted).
+
+`indices.merge.disk.watermark.high.max_headroom`
+:   Controls the max headroom for the merge disk usage watermark, in case it is specified as percentage or ratio values.
+Defaults to `100GB` when `indices.merge.disk.watermark.high` is not explicitly set.
+This caps the amount of free disk space before merge scheduling is blocked.
 

docs/reference/modules/threadpool.asciidoc

@@ -79,10 +79,13 @@ There are several thread pools, but the important ones include:
     default maximum size of `min(5, (`<<node.processors,
     `# of allocated processors`>>`) / 2)`.
 
+`merge`::
+    For [merge](https://www.elastic.co/guide/en/elasticsearch/reference/current/index-modules-merge.html) operations of all the shards on the node.
+    Thread pool type is `scaling` with a keep-alive of `5m` and a default maximum size of [`# of allocated processors`](#node.processors).
+
 `force_merge`::
-    For <<indices-forcemerge,force merge>> operations.
-    Thread pool type is `fixed` with a size of `max(1, (`<<node.processors,
-`# of allocated processors`>>`) / 8)` and an unbounded queue size.
+    For waiting on blocking [force merge](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-indices-forcemerge) operations.
+    Thread pool type is `fixed` with a size of `max(1, (`[`# of allocated processors`](#node.processors)`) / 8)` and an unbounded queue size.
 
 `management`::
     For cluster management.

server/src/main/java/org/elasticsearch/common/util/Result.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.common.util;
+
+import org.elasticsearch.common.CheckedSupplier;
+
+import java.util.Optional;
+
+/**
+ * A wrapper around either
+ * <ul>
+ * <li>a successful result of parameterized type {@code V}</li>
+ * <li>a failure with exception type {@code E}</li>
+ * </ul>
+ */
+public abstract class Result<V, E extends Exception> implements CheckedSupplier<V, E> {
+
+    public static <V, E extends Exception> Result<V, E> of(V value) {
+        return new Success<>(value);
+    }
+
+    public static <V, E extends Exception> Result<V, E> failure(E exception) {
+        return new Failure<>(exception);
+    }
+
+    private Result() {}
+
+    public abstract V get() throws E;
+
+    public abstract Optional<E> failure();
+
+    public abstract boolean isSuccessful();
+
+    public boolean isFailure() {
+        return isSuccessful() == false;
+    };
+
+    public abstract Optional<V> asOptional();
+
+    private static class Success<V, E extends Exception> extends Result<V, E> {
+        private final V value;
+
+        Success(V value) {
+            this.value = value;
+        }
+
+        @Override
+        public V get() throws E {
+            return value;
+        }
+
+        @Override
+        public Optional<E> failure() {
+            return Optional.empty();
+        }
+
+        @Override
+        public boolean isSuccessful() {
+            return true;
+        }
+
+        @Override
+        public Optional<V> asOptional() {
+            return Optional.of(value);
+        }
+    }
+
+    private static class Failure<V, E extends Exception> extends Result<V, E> {
+        private final E exception;
+
+        Failure(E exception) {
+            this.exception = exception;
+        }
+
+        @Override
+        public V get() throws E {
+            throw exception;
+        }
+
+        @Override
+        public Optional<E> failure() {
+            return Optional.of(exception);
+        }
+
+        @Override
+        public boolean isSuccessful() {
+            return false;
+        }
+
+        @Override
+        public Optional<V> asOptional() {
+            return Optional.empty();
+        }
+    }
+}

server/src/test/java/org/elasticsearch/common/util/ResultTests.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.common.util;
+
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.ElasticsearchStatusException;
+import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.test.ESTestCase;
+
+import static org.elasticsearch.test.hamcrest.OptionalMatchers.isEmpty;
+import static org.elasticsearch.test.hamcrest.OptionalMatchers.isPresentWith;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.sameInstance;
+
+public class ResultTests extends ESTestCase {
+
+    public void testSuccess() {
+        final String str = randomAlphaOfLengthBetween(3, 8);
+        final Result<String, ElasticsearchException> result = Result.of(str);
+        assertThat(result.isSuccessful(), is(true));
+        assertThat(result.isFailure(), is(false));
+        assertThat(result.get(), sameInstance(str));
+        assertThat(result.failure(), isEmpty());
+        assertThat(result.asOptional(), isPresentWith(str));
+    }
+
+    public void testFailure() {
+        final ElasticsearchException exception = new ElasticsearchStatusException(
+            randomAlphaOfLengthBetween(10, 30),
+            RestStatus.INTERNAL_SERVER_ERROR
+        );
+        final Result<String, ElasticsearchException> result = Result.failure(exception);
+        assertThat(result.isSuccessful(), is(false));
+        assertThat(result.isFailure(), is(true));
+        assertThat(expectThrows(Exception.class, result::get), sameInstance(exception));
+        assertThat(result.failure(), isPresentWith(sameInstance(exception)));
+        assertThat(result.asOptional(), isEmpty());
+    }
+
+}