Document and test operator-only node bandwidth recovery settings (#83372)

This commit updates the Operator-only functionality doc to 
mention the operator only settings introduced in #82819.

It also adds an integration test for those operator only 
settings that would have caught #83359.

docs/reference/modules/indices/recovery.asciidoc

@@ -109,6 +109,7 @@ Do not increase this setting without carefully verifying that your cluster has
 the resources available to handle the extra load that will result.
 
 [discrete]
+[[recovery-settings-for-managed-services]]
 ==== Recovery settings for managed services
 
 NOTE: {cloud-only}

x-pack/docs/en/security/operator-privileges/operator-only-functionality.asciidoc

@@ -30,7 +30,7 @@ given {es} version.
 ==== Operator-only dynamic cluster settings
 
 * All <<ip-filtering,IP filtering>> settings
-* The following the dynamic <<ml-settings,machine learning settings>>:
+* The following dynamic <<ml-settings,machine learning settings>>:
   - `xpack.ml.node_concurrent_job_allocations`
   - `xpack.ml.max_machine_memory_percent`
   - `xpack.ml.use_auto_machine_memory_percent`
@@ -41,3 +41,9 @@ given {es} version.
   - `xpack.ml.enable_config_migration`
   - `xpack.ml.persist_results_max_retries`
 * The <<cluster-routing-disk-threshold,`cluster.routing.allocation.disk.threshold_enabled` setting>>
+* The following <<recovery-settings-for-managed-services,recovery settings for managed services>>:
+  - `node.bandwidth.recovery.operator.factor`
+  - `node.bandwidth.recovery.operator.factor.read`
+  - `node.bandwidth.recovery.operator.factor.write`
+  - `node.bandwidth.recovery.operator.factor.max_overcommit`
+

x-pack/plugin/security/qa/operator-privileges-tests/src/javaRestTest/java/org/elasticsearch/xpack/security/operator/OperatorPrivilegesIT.java

@@ -14,6 +14,7 @@ import org.elasticsearch.common.settings.SecureString;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.util.set.Sets;
+import org.elasticsearch.indices.recovery.RecoverySettings;
 import org.elasticsearch.test.rest.ESRestTestCase;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.json.JsonXContent;
@@ -22,6 +23,7 @@ import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.Base64;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -31,6 +33,7 @@ import java.util.stream.Collectors;
 
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
 
 public class OperatorPrivilegesIT extends ESRestTestCase {
 
@@ -203,6 +206,48 @@ public class OperatorPrivilegesIT extends ESRestTestCase {
         assertNull(persistentSettings.get("search.allow_expensive_queries"));
     }
 
+    public void testNodeBandwidthRecoveryUserFactorSettings() throws IOException {
+        final Map<String, Double> userSettings = Map.of(
+            RecoverySettings.NODE_BANDWIDTH_RECOVERY_FACTOR_READ_SETTING.getKey(),
+            randomDoubleBetween(0.1d, 1.0d, true),
+            RecoverySettings.NODE_BANDWIDTH_RECOVERY_FACTOR_WRITE_SETTING.getKey(),
+            randomDoubleBetween(0.1d, 1.0d, true)
+        );
+
+        for (Map.Entry<String, Double> userSetting : userSettings.entrySet()) {
+            updateSettings(Map.of(userSetting.getKey(), userSetting.getValue()), null);
+            assertThat(getPersistentSettings().get(userSetting.getKey()), equalTo(Double.toString(userSetting.getValue())));
+
+            updateSettings(Collections.singletonMap(userSetting.getKey(), null), OPERATOR_AUTH_HEADER);
+            assertThat(getPersistentSettings().containsKey(userSetting.getKey()), is(false));
+        }
+    }
+
+    public void testNodeBandwidthRecoveryOperatorFactorSettings() throws IOException {
+        final Map<String, Double> operatorSettings = Map.of(
+            RecoverySettings.NODE_BANDWIDTH_RECOVERY_OPERATOR_FACTOR_SETTING.getKey(),
+            randomDoubleBetween(0.1d, 1.0d, true),
+            RecoverySettings.NODE_BANDWIDTH_RECOVERY_OPERATOR_FACTOR_READ_SETTING.getKey(),
+            randomDoubleBetween(0.1d, 1.0d, true),
+            RecoverySettings.NODE_BANDWIDTH_RECOVERY_OPERATOR_FACTOR_WRITE_SETTING.getKey(),
+            randomDoubleBetween(0.1d, 1.0d, true),
+            RecoverySettings.NODE_BANDWIDTH_RECOVERY_OPERATOR_FACTOR_MAX_OVERCOMMIT_SETTING.getKey(),
+            randomDoubleBetween(1d, 1000d, true)
+        );
+
+        for (Map.Entry<String, Double> operatorSetting : operatorSettings.entrySet()) {
+            final Map<String, Double> settings = Map.of(operatorSetting.getKey(), operatorSetting.getValue());
+
+            ResponseException exception = expectThrows(ResponseException.class, () -> updateSettings(settings, null));
+            assertThat(exception.getResponse().getStatusLine().getStatusCode(), equalTo(403));
+            assertThat(exception.getMessage(), containsString("is unauthorized for user"));
+            assertThat(getPersistentSettings().containsKey(operatorSetting.getKey()), is(false));
+
+            updateSettings(settings, OPERATOR_AUTH_HEADER);
+            assertThat(getPersistentSettings().get(operatorSetting.getKey()), equalTo(Double.toString(operatorSetting.getValue())));
+        }
+    }
+
     private void createSnapshotRepo(String repoName) throws IOException {
         Request request = new Request("PUT", "/_snapshot/" + repoName);
         request.setJsonEntity(