Account for replicated data streams more completely.

Author: jbaiera

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

@@ -214,7 +214,12 @@ public DataStream(
             lifecycle,
             dataStreamOptions,
             new DataStreamIndices(BACKING_INDEX_PREFIX, List.copyOf(indices), rolloverOnWrite, autoShardingEvent),
-            new DataStreamIndices(FAILURE_STORE_PREFIX, List.copyOf(failureIndices), failureIndices.isEmpty(), null)
+            new DataStreamIndices(
+                FAILURE_STORE_PREFIX,
+                List.copyOf(failureIndices),
+                (replicated == false && failureIndices.isEmpty()),
+                null
+            )
         );
     }
 
@@ -283,15 +288,15 @@ public static DataStream read(StreamInput in) throws IOException {
             backingIndicesBuilder.setAutoShardingEvent(in.readOptionalWriteable(DataStreamAutoShardingEvent::new));
         }
         if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_15_0)) {
-            // Read the rollover on write flag from the stream, but force it on if the failure indices are empty
-            boolean failureStoreRolloverOnWrite = in.readBoolean();
-            failureStoreRolloverOnWrite |= failureIndices.isEmpty();
+            // Read the rollover on write flag from the stream, but force it on if the failure indices are empty and we're not replicating
+            boolean failureStoreRolloverOnWrite = in.readBoolean() || (replicated == false && failureIndices.isEmpty());
             failureIndicesBuilder.setRolloverOnWrite(failureStoreRolloverOnWrite)
                 .setAutoShardingEvent(in.readOptionalWriteable(DataStreamAutoShardingEvent::new));
         } else {
             // If we are reading from an older version that does not have these fields, just default
             // to a reasonable value for rollover on write for the failure store
-            failureIndicesBuilder.setRolloverOnWrite(failureIndices.isEmpty());
+            boolean failureStoreRolloverOnWrite = replicated == false && failureIndices.isEmpty();
+            failureIndicesBuilder.setRolloverOnWrite(failureStoreRolloverOnWrite);
         }
         DataStreamOptions dataStreamOptions;
         if (in.getTransportVersion().onOrAfter(TransportVersions.V_8_16_0)) {
@@ -1497,9 +1502,11 @@ public void writeTo(StreamOutput out) throws IOException {
             new DataStreamIndices(
                 FAILURE_STORE_PREFIX,
                 args[13] != null ? (List<Index>) args[13] : List.of(),
-                // We check if the list of failure indices are empty first, forcing rollover on write to true, and if they have entries,
-                // then we use whatever value was previously present.
-                (args[13] != null && ((List<Index>) args[13]).isEmpty()) || (args[14] != null && (boolean) args[14]),
+                // If replicated (args[5]) is null or exists and is false, and the failure index list (args[13]) is null or
+                // exists and is empty, then force the rollover on write field to true. If none of those conditions are met,
+                // then use the rollover on write value (args[14]) present in the parser.
+                ((args[5] == null || ((boolean) args[5] == false)) && (args[13] == null || ((List<Index>) args[13]).isEmpty()))
+                    || (args[14] != null && (boolean) args[14]),
                 (DataStreamAutoShardingEvent) args[15]
             )
         )

test/framework/src/main/java/org/elasticsearch/cluster/metadata/DataStreamTestHelper.java

@@ -391,7 +391,7 @@ public static DataStream randomInstance(String dataStreamName, LongSupplier time
                 )
                 .build(),
             DataStream.DataStreamIndices.failureIndicesBuilder(failureIndices)
-                .setRolloverOnWrite(failureStore && replicated == false && randomBoolean())
+                .setRolloverOnWrite(replicated == false && (failureIndices.isEmpty() || randomBoolean()))
                 .setAutoShardingEvent(
                     failureStore && randomBoolean()
                         ? new DataStreamAutoShardingEvent(