[8.19] Add existing shards allocator settings to failure store allowed list. (#131056)

docs/changelog/131056.yaml

@@ -0,0 +1,5 @@
+pr: 131056
+summary: Add existing shards allocator settings to failure store allowed list
+area: Data streams
+type: bug
+issues: []

server/src/main/java/org/elasticsearch/cluster/metadata/DataStreamFailureStoreDefinition.java

@@ -10,6 +10,7 @@
 package org.elasticsearch.cluster.metadata;
 
 import org.elasticsearch.cluster.routing.allocation.DataTier;
+import org.elasticsearch.cluster.routing.allocation.ExistingShardsAllocator;
 import org.elasticsearch.common.compress.CompressedXContent;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
@@ -40,7 +41,9 @@ public class DataStreamFailureStoreDefinition {
         IndexMetadata.SETTING_NUMBER_OF_SHARDS,
         IndexMetadata.SETTING_NUMBER_OF_REPLICAS,
         IndexMetadata.SETTING_AUTO_EXPAND_REPLICAS,
-        IndexSettings.INDEX_REFRESH_INTERVAL_SETTING.getKey()
+        IndexSettings.INDEX_REFRESH_INTERVAL_SETTING.getKey(),
+        // Different recovery implementations may be provided on the index which need to be preserved.
+        ExistingShardsAllocator.EXISTING_SHARDS_ALLOCATOR_SETTING.getKey()
     );
     public static final Set<String> SUPPORTED_USER_SETTINGS_PREFIXES = Set.of(
         IndexMetadata.INDEX_ROUTING_REQUIRE_GROUP_PREFIX + ".",

server/src/test/java/org/elasticsearch/cluster/metadata/MetadataCreateIndexServiceTests.java

@@ -30,6 +30,7 @@
 import org.elasticsearch.cluster.routing.RoutingTable;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.routing.allocation.DataTier;
+import org.elasticsearch.cluster.routing.allocation.ExistingShardsAllocator;
 import org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator;
 import org.elasticsearch.cluster.routing.allocation.decider.AllocationDeciders;
 import org.elasticsearch.cluster.routing.allocation.decider.MaxRetryAllocationDecider;
@@ -735,6 +736,55 @@ public boolean overrulesTemplateAndRequestSettings() {
         assertThat(aggregatedIndexSettings.get("other_setting"), equalTo("other_value"));
     }
 
+    /**
+     * When a failure store index is created, we must filter out any unsupported settings from the create request or from the template that
+     * may have been provided by users in the create request or from the original data stream template. An exception to this is any settings
+     * that have been provided by index setting providers which should be considered default values on indices.
+     */
+    public void testAggregateSettingsProviderIsNotFilteredOnFailureStore() {
+        IndexTemplateMetadata templateMetadata = addMatchingTemplate(builder -> {
+            builder.settings(Settings.builder().put("template_setting", "value1"));
+        });
+        ProjectMetadata projectMetadata = ProjectMetadata.builder(projectId).templates(Map.of("template_1", templateMetadata)).build();
+        ClusterState clusterState = ClusterState.builder(ClusterName.DEFAULT).putProjectMetadata(projectMetadata).build();
+        var request = new CreateIndexClusterStateUpdateRequest("create index", projectId, "test", "test").settings(
+            Settings.builder().put("request_setting", "value2").build()
+        ).isFailureIndex(true);
+
+        Settings aggregatedIndexSettings = aggregateIndexSettings(
+            clusterState,
+            request,
+            templateMetadata.settings(),
+            null,
+            null,
+            Settings.EMPTY,
+            IndexScopedSettings.DEFAULT_SCOPED_SETTINGS,
+            randomShardLimitService(),
+            Set.of(new IndexSettingProvider() {
+                @Override
+                public Settings getAdditionalIndexSettings(
+                    String indexName,
+                    String dataStreamName,
+                    IndexMode templateIndexMode,
+                    ProjectMetadata projectMetadata,
+                    Instant resolvedAt,
+                    Settings indexTemplateAndCreateRequestSettings,
+                    List<CompressedXContent> combinedTemplateMappings
+                ) {
+                    return Settings.builder().put(ExistingShardsAllocator.EXISTING_SHARDS_ALLOCATOR_SETTING.getKey(), "override").build();
+                }
+
+                @Override
+                public boolean overrulesTemplateAndRequestSettings() {
+                    return true;
+                }
+            })
+        );
+
+        assertThat(aggregatedIndexSettings.get("template_setting"), nullValue());
+        assertThat(aggregatedIndexSettings.get(ExistingShardsAllocator.EXISTING_SHARDS_ALLOCATOR_SETTING.getKey()), equalTo("override"));
+    }
+
     public void testAggregateSettingsProviderOverrulesNullFromRequest() {
         IndexTemplateMetadata templateMetadata = addMatchingTemplate(builder -> {
             builder.settings(Settings.builder().put("template_setting", "value1"));