Update modification behavior

modules/data-streams/src/yamlRestTest/resources/rest-api-spec/test/data_stream/170_modify_data_stream.yml

@@ -211,17 +211,6 @@
                 index: test_index2
                 failure_store: true
 
-  # We are not allowed to remove the write index for the failure store
-  - do:
-      catch: /cannot remove backing index \[.*\] of data stream \[data-stream-for-modification\] because it is the write index/
-      indices.modify_data_stream:
-        body:
-          actions:
-            - remove_backing_index:
-                data_stream: "data-stream-for-modification"
-                index: $write_failure_index
-                failure_store: true
-
   # We will not accept an index that is already part of the data stream's backing indices
   - do:
       catch: /cannot add index \[.*\] to data stream \[data-stream-for-modification\] because it is already a backing index on data stream \[data-stream-for-modification\]/
@@ -298,6 +287,105 @@
   - match: { data_streams.0.failure_store.indices.0.index_name: $first_failure_index }
   - match: { data_streams.0.failure_store.indices.1.index_name: $write_failure_index }
 
+  # Remove write index of the failure store
+  - do:
+      indices.modify_data_stream:
+        body:
+          actions:
+            - remove_backing_index:
+                data_stream: "data-stream-for-modification"
+                index: $write_failure_index
+                failure_store: true
+  - is_true: acknowledged
+
+  - do:
+      indices.get_data_stream:
+        name: "data-stream-for-modification"
+  - match: { data_streams.0.name: data-stream-for-modification }
+  - match: { data_streams.0.timestamp_field.name: '@timestamp' }
+  - match: { data_streams.0.generation: 6 }
+  - length: { data_streams.0.indices: 1 }
+  - length: { data_streams.0.failure_store.indices: 1 }
+  - match: { data_streams.0.indices.0.index_name: $write_index }
+  - match: { data_streams.0.failure_store.indices.0.index_name: $first_failure_index }
+
+  # Remove the last write index of the failure store
+  - do:
+      indices.modify_data_stream:
+        body:
+          actions:
+            - remove_backing_index:
+                data_stream: "data-stream-for-modification"
+                index: $first_failure_index
+                failure_store: true
+  - is_true: acknowledged
+
+  - do:
+      indices.get_data_stream:
+        name: "data-stream-for-modification"
+  - match: { data_streams.0.name: data-stream-for-modification }
+  - match: { data_streams.0.timestamp_field.name: '@timestamp' }
+  - match: { data_streams.0.generation: 7 }
+  - length: { data_streams.0.indices: 1 }
+  - length: { data_streams.0.failure_store.indices: 0 }
+  - match: { data_streams.0.indices.0.index_name: $write_index }
+
+  # Doing these checks again to make sure we still return the same error with  an empty failure store
+  # We will not accept an index that is already part of the data stream's backing indices
+  - do:
+      catch: /cannot add index \[.*\] to data stream \[data-stream-for-modification\] because it is already a backing index on data stream \[data-stream-for-modification\]/
+      indices.modify_data_stream:
+        body:
+          actions:
+            - add_backing_index:
+                data_stream: "data-stream-for-modification"
+                index: $write_index
+                failure_store: true
+
+  # We will not accept an index that is already part of a different data stream's backing indices
+  - do:
+      catch: /cannot add index \[.*\] to data stream \[data-stream-for-modification\] because it is already a backing index on data stream \[data-stream-for-modification2\]/
+      indices.modify_data_stream:
+        body:
+          actions:
+            - add_backing_index:
+                data_stream: "data-stream-for-modification"
+                index: $second_write_index
+                failure_store: true
+
+  # We will not accept an index that is already part of a different data stream's failure store
+  - do:
+      catch: /cannot add index \[.*\] to data stream \[data-stream-for-modification\] because it is already a failure store index on data stream \[data-stream-for-modification2\]/
+      indices.modify_data_stream:
+        body:
+          actions:
+            - add_backing_index:
+                data_stream: "data-stream-for-modification"
+                index: $second_write_failure_index
+                failure_store: true
+
+  # We will return a failed response if we try to remove an index from the failure store that is not present
+  - do:
+      catch: /index \[.*\] not found/
+      indices.modify_data_stream:
+        body:
+          actions:
+            - remove_backing_index:
+                data_stream: "data-stream-for-modification"
+                index: $write_index
+                failure_store: true
+
+  # Add index to empty failure store
+  - do:
+      indices.modify_data_stream:
+        body:
+          actions:
+            - add_backing_index:
+                data_stream: "data-stream-for-modification"
+                index: "test_index1"
+                failure_store: true
+  - is_true: acknowledged
+
   - do:
       indices.delete_data_stream:
         name: data-stream-for-modification

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

@@ -581,23 +581,13 @@ public DataStream removeFailureStoreIndex(Index index) {
             );
         }
 
-        // TODO: When failure stores are lazily created, this wont necessarily be required anymore. We can remove the failure store write
-        // index as long as we mark the data stream to lazily rollover the failure store with no conditions on its next write
-        if (failureIndices.indices.size() == (failureIndexPosition + 1)) {
-            throw new IllegalArgumentException(
-                String.format(
-                    Locale.ROOT,
-                    "cannot remove backing index [%s] of data stream [%s] because it is the write index of the failure store",
-                    index.getName(),
-                    name
-                )
-            );
-        }
-
+        // If this is the write index, we're marking the failure store for lazy rollover, to make sure a new write index gets created on the
+        // next write. We do this regardless of whether it's the last index in the failure store or not.
+        boolean rolloverOnWrite = failureIndices.indices.size() == (failureIndexPosition + 1);
         List<Index> updatedFailureIndices = new ArrayList<>(failureIndices.indices);
         updatedFailureIndices.remove(index);
         assert updatedFailureIndices.size() == failureIndices.indices.size() - 1;
-        return copy().setFailureIndices(failureIndices.copy().setIndices(updatedFailureIndices).build())
+        return copy().setFailureIndices(failureIndices.copy().setIndices(updatedFailureIndices).setRolloverOnWrite(rolloverOnWrite).build())
             .setGeneration(generation + 1)
             .build();
     }